feat: 线程操作与 skills/MCP 管理面暴露（host CLI + 服务端 API + 控制台）

线程：host.py 新增 thread-list/read/rename/archive/unarchive/compact 子命令（纯 JSON stdout 契约，read/rename/unarchive/compact 先 resume）；服务端按任务定 scope（thread/list 的 cwd 过滤解析，Docker/cloud/异机 409，改操作仅终态，审计 thread_*）；控制台线程区（元数据、列表、重命名、归档、取消归档、压缩，终态门控）。

skills/MCP：host.py 新增 skills-list/apps-list/mcp-servers-list/mcp-servers-reload/mcp-server-oauth-login 子命令；服务端按项目 scope（读 read，改 manage_project，审计 mcp_*）；控制台技能与 MCP 区（列表、重载、OAuth 发起展示，绝不谎称登录完成）。

测试：host 84、cloud 687（4 跳过）、deploy 19 全过。诚实边界：Docker 任务线程在容器内故 409；无仓库任务推不出 cwd 时 404；OAuth 只做到发起透传，完成态靠协议通知；merged/merge-failed 改操作 409（比控制台门控窄，服务端报错原文展示）。

cloud/console.html

@@ -150,6 +150,31 @@
 </div>
 </div>
 
+<section class="card">
+<h2>技能与 MCP</h2>
+<div class="muted">反映宿主租户 home，容器内 home 可能不同；重载只在 schema 枚举里，当前 pin 的二进制拒绝时原样 502；OAuth 只发起登录，不代表登录完成。</div>
+<div class="bar">
+  <button id="refreshSkills">刷新</button>
+  <span id="skillMsg" class="muted">先选项目</span>
+</div>
+<h3>技能</h3>
+<div id="skillList" class="muted">先选项目</div>
+<h3>应用</h3>
+<div id="appList" class="muted">先选项目</div>
+<h3>MCP 服务</h3>
+<div id="mcpList" class="muted">先选项目</div>
+<div class="bar">
+  <button id="btnReloadMcp">重载配置</button>
+  <span id="mcpMsg" class="muted"></span>
+</div>
+<div class="bar">
+  <input id="mcpOauthName" type="text" placeholder="MCP 服务名">
+  <button id="btnMcpOauth">发起 OAuth 登录</button>
+  <span id="mcpOauthMsg" class="muted"></span>
+</div>
+<div id="mcpOauthBox" class="muted"></div>
+</section>
+
 <section class="card">
 <h2>环境 <span class="muted">（选、建、管：当前项目的基准检出）</span></h2>
 <div class="bar">
@@ -379,6 +404,19 @@
 <h3>分诊收件箱</h3>
 <div class="bar"><button id="btnLoadTriage">刷新分诊</button><span id="triageMsg" class="muted"></span></div>
 <div id="triageBox" class="muted"></div>
+<h3>线程</h3>
+<div id="threadBox" class="muted">点任务行查看详情后，这里显示线程元数据</div>
+<div class="bar" id="threadActions">
+  <input id="threadRenameInput" type="text" placeholder="新线程名">
+  <button id="btnThreadRename">重命名</button>
+  <button id="btnThreadArchive">归档</button>
+  <button id="btnThreadUnarchive">取消归档</button>
+  <button id="btnThreadCompact">压缩</button>
+  <span id="threadGateNote" class="muted"></span>
+  <span id="threadMsg" class="muted"></span>
+</div>
+<div id="threadList" class="muted"></div>
+<div class="muted">限制：Docker/异机任务线程操作一律 409；merged/merge-failed 不可改；turn/steer 与 review/start 不在本批。</div>
 </section>
 
 <script>
@@ -390,6 +428,7 @@ var traceAbort = null;
 var TRACE_KIND = { exec: "命令", file: "文件", call: "调用", tool: "工具", message: "消息", reasoning: "思考", plan: "计划", "agent-start": "子代理", "agent-done": "子代理" };
 var TRACE_LIMIT = 1000;
 var TRACE_TERMINAL = ["completed", "failed", "interrupted", "merged", "merge-failed"];
+var THREAD_MUTATION_STATUSES = ["completed", "failed", "interrupted"];
 
 function esc(s) {
   return String(s == null ? "" : s).replace(/&/g, "&amp;").replace(/</g, "&lt;")
@@ -1018,6 +1057,11 @@ async function showDetail(taskId) {
     loadFollowups(taskId);
     loadPeerMessages(taskId);
     $("#peerMsg").textContent = "";
+    $("#threadMsg").textContent = "";
+    $("#threadRenameInput").value = "";
+    setThreadGate(t.status);
+    loadThread(taskId);
+    loadThreads(taskId);
   } catch (e) { $("#detail").innerHTML = "<span class='err'>加载失败：" + esc(e.message) + "</span>"; }
 }
 
@@ -1279,7 +1323,10 @@ $("#refreshProjects").onclick = loadProjects;
 $("#refreshWorkers").onclick = loadWorkers;
 $("#refreshTasks").onclick = loadTasks;
 $("#refreshEnvironments").onclick = loadEnvironments;
-$("#projectSelect").onchange = function () { loadTasks(); loadEnvironments(); loadMembers(); };
+// 合并取舍（P2 rebase 到 97f019d）：两侧全留——项目下拉切换时
+// 同时刷新任务、环境、成员与项目技能；成员、筛选、链路、管理
+// 绑定与技能加载均保留，任一侧都不丢。
+$("#projectSelect").onchange = function () { loadTasks(); loadEnvironments(); loadMembers(); loadProjectSkills(); };
 $("#refreshMembers").onclick = loadMembers;
 $("#inviteMember").onclick = inviteMember;
 $("#taskStatusFilter").onchange = renderTasks;
@@ -1500,10 +1547,130 @@ async function postPeerMessage(mode) {
   } catch (e) { $("#peerMsg").innerHTML = "<span class='err'>发送失败：" + esc(e.message) + "</span>"; }
 }
 
+async function loadThread(taskId) {
+  var box = $("#threadBox");
+  try {
+    var d = await api("/tasks/" + encodeURIComponent(taskId) + "/thread", "GET");
+    var th = d.thread || d;
+    var turns = th.turns;
+    if (turns === undefined || turns === null) turns = th.turn_count;
+    if (turns === undefined || turns === null) turns = th.turns_count;
+    box.innerHTML =
+      "id " + esc(th.id || taskId) +
+      " | 名字 " + esc(th.name || th.thread_name || "") +
+      " | 归档 " + esc(th.archived === undefined || th.archived === null ? "" : (th.archived ? "是" : "否")) +
+      " | cwd " + esc(th.cwd || "") +
+      (turns !== undefined && turns !== null ? " | turn 数 " + esc(turns) : "");
+  } catch (e) { box.innerHTML = "<span class='err'>线程加载失败：" + esc(e.message) + "</span>"; }
+}
+async function loadThreads(taskId) {
+  var box = $("#threadList");
+  try {
+    var d = await api("/tasks/" + encodeURIComponent(taskId) + "/threads", "GET");
+    var list = d.threads || [];
+    if (!list.length) { box.innerHTML = "暂无其他线程"; return; }
+    box.innerHTML = list.map(function (t) {
+      return esc(t.name || t.thread_name || t.id || "") + " [" + esc(t.id || "") + "]" +
+        (t.archived ? " [已归档]" : "") + (t.cwd ? " " + esc(t.cwd) : "");
+    }).join("<br>");
+  } catch (e) { box.innerHTML = "<span class='err'>线程列表加载失败：" + esc(e.message) + "</span>"; }
+}
+function setThreadGate(status) {
+  var ok = THREAD_MUTATION_STATUSES.indexOf(status) >= 0;
+  ["btnThreadRename", "btnThreadArchive", "btnThreadUnarchive", "btnThreadCompact"].forEach(function (id) {
+    var b = $("#" + id);
+    if (b) b.disabled = !ok;
+  });
+  $("#threadGateNote").textContent = ok ? "终态任务，线程操作可用" : "仅终态任务可操作线程，当前状态 " + status + "，按钮已置灰";
+}
+async function postThreadAction(kind) {
+  if (!currentTask) return;
+  if (THREAD_MUTATION_STATUSES.indexOf(currentTask.status) < 0) {
+    $("#threadMsg").textContent = "仅 completed/failed/interrupted 任务可操作线程（merged/merge-failed 不可改，服务端同样 409）";
+    return;
+  }
+  var body = {};
+  if (kind === "rename") {
+    var name = $("#threadRenameInput").value.trim();
+    if (!name) { $("#threadMsg").textContent = "新线程名不能为空"; return; }
+    body = { name: name };
+  }
+  try {
+    var threadPaths = { rename: "/thread/rename", archive: "/thread/archive", unarchive: "/thread/unarchive", compact: "/thread/compact" };
+    var r = await api("/tasks/" + encodeURIComponent(currentTask.id) + threadPaths[kind], "POST", body);
+    $("#threadMsg").innerHTML = "<span class='ok'>" + kind + " 已提交：" + esc(JSON.stringify(r).slice(0, 300)) + "</span>";
+    loadThread(currentTask.id);
+    loadThreads(currentTask.id);
+  } catch (e) { $("#threadMsg").innerHTML = "<span class='err'>" + kind + " 失败：" + esc(e.message) + "</span>"; }
+}
+async function loadProjectSkills() {
+  var pid = $("#projectSelect").value;
+  if (!pid) {
+    $("#skillList").textContent = "先选项目";
+    $("#appList").textContent = "先选项目";
+    $("#mcpList").textContent = "先选项目";
+    $("#skillMsg").textContent = "先选项目";
+    return;
+  }
+  $("#skillMsg").textContent = "";
+  try {
+    var s = await api("/projects/" + encodeURIComponent(pid) + "/skills", "GET");
+    var skills = s.skills || [];
+    $("#skillList").innerHTML = skills.length ? skills.map(function (x) {
+      return esc(x.name || x.id || "") + (x.description ? " - " + esc(x.description) : "") + (x.detail ? " - " + esc(x.detail) : "");
+    }).join("<br>") : "暂无技能";
+  } catch (e) { $("#skillList").innerHTML = "<span class='err'>技能加载失败：" + esc(e.message) + "</span>"; }
+  try {
+    var a = await api("/projects/" + encodeURIComponent(pid) + "/apps", "GET");
+    var apps = a.apps || [];
+    $("#appList").innerHTML = apps.length ? apps.map(function (x) {
+      return esc(x.name || x.id || "") + (x.description ? " - " + esc(x.description) : "");
+    }).join("<br>") : "暂无应用";
+  } catch (e) { $("#appList").innerHTML = "<span class='err'>应用加载失败：" + esc(e.message) + "</span>"; }
+  try {
+    var m = await api("/projects/" + encodeURIComponent(pid) + "/mcp-servers", "GET");
+    var servers = m.servers || m.mcp_servers || [];
+    $("#mcpList").innerHTML = servers.length ? servers.map(function (x) {
+      var det = x.detail || x.details || x.state || "";
+      return esc(x.name || x.id || "") + " | 状态 " + esc(x.status || "") + (det ? " | " + esc(det) : "");
+    }).join("<br>") : "暂无 MCP 服务";
+  } catch (e) { $("#mcpList").innerHTML = "<span class='err'>MCP 服务加载失败：" + esc(e.message) + "</span>"; }
+}
 $("#btnQueueFollowup").onclick = function () { postFollowup("queued"); };
 $("#btnDirectFollowup").onclick = function () { postFollowup("direct"); };
 $("#btnQueuePeerMessage").onclick = function () { postPeerMessage("queued"); };
 $("#btnDirectPeerMessage").onclick = function () { postPeerMessage("direct"); };
+$("#btnThreadRename").onclick = function () { postThreadAction("rename"); };
+$("#btnThreadArchive").onclick = function () { postThreadAction("archive"); };
+$("#btnThreadUnarchive").onclick = function () { postThreadAction("unarchive"); };
+$("#btnThreadCompact").onclick = function () { postThreadAction("compact"); };
+$("#refreshSkills").onclick = loadProjectSkills;
+$("#btnReloadMcp").onclick = async function () {
+  var pid = $("#projectSelect").value;
+  if (!pid) { $("#mcpMsg").textContent = "先选项目"; return; }
+  try {
+    await api("/projects/" + encodeURIComponent(pid) + "/mcp-servers/reload", "POST", {});
+    $("#mcpMsg").innerHTML = "<span class='ok'>配置已重载</span>";
+    loadProjectSkills();
+  } catch (e) { $("#mcpMsg").innerHTML = "<span class='err'>重载失败：" + esc(e.message) + "</span>"; }
+};
+$("#btnMcpOauth").onclick = async function () {
+  var pid = $("#projectSelect").value;
+  var name = $("#mcpOauthName").value.trim();
+  if (!pid) { $("#mcpOauthMsg").textContent = "先选项目"; return; }
+  if (!name) { $("#mcpOauthMsg").textContent = "服务名不能为空"; return; }
+  try {
+    var d = await api("/projects/" + encodeURIComponent(pid) + "/mcp-servers/" + encodeURIComponent(name) + "/oauth-login", "POST", {});
+    var url = d.url || d.auth_url || d.login_url || d.authorization_url || "";
+    var steps = d.steps || d.instructions || "";
+    var html = "登录尚未完成：请在浏览器打开下面的链接或按步骤完成，完成后回到本页刷新 MCP 状态。<br>";
+    if (url) html += "<a href='" + esc(url) + "' target='_blank' rel='noopener'>" + esc(url) + "</a><br>";
+    if (steps) html += esc(Array.isArray(steps) ? steps.join("\n") : String(steps)) + "<br>";
+    html += "<pre>" + esc(JSON.stringify(d, null, 2).slice(0, 4000)) + "</pre>";
+    $("#mcpOauthBox").innerHTML = html;
+    $("#mcpOauthMsg").innerHTML = "<span class='ok'>已返回登录信息，见下方</span>";
+  } catch (e) { $("#mcpOauthMsg").innerHTML = "<span class='err'>发起失败：" + esc(e.message) + "</span>"; }
+};
 $("#btnCost").onclick = async function () {
   if (!currentTask) return;
   try {

cloud/server.py

@@ -46,6 +46,8 @@ import re
 import shutil
 import socket
 import sqlite3
+import subprocess
+import sys
 import threading
 import time
 import uuid
@@ -165,6 +167,12 @@ _server_log = logging.getLogger("cloud.server")
 TERMINAL_STATUSES = frozenset(
     {"completed", "failed", "interrupted", "merged", "merge-failed"}
 )
+# 线程改操作（rename/archive/unarchive/compact）只允许这三个终态：
+# 比 TERMINAL_STATUSES 窄——merged/merge-failed 的 thread 已随任务
+# 终局，不再改；running/queued 一律 409。
+THREAD_MUTATION_STATUSES = frozenset(
+    {"completed", "failed", "interrupted"}
+)
 # 实时链路（任务过程事件流）：host 在回合进行中把 app-server
 # 通知摘要成条目追加写 jsonl，服务端 tail 文件经 SSE 推送。
 TRACE_STREAM_MAX_SECONDS = 3600
@@ -254,6 +262,211 @@ def _trace_dirs_for_task(server, task):
     return dirs
 
 
+# -- 线程 / skills / MCP 元数据操作（经 host.py CLI 透传）--------------
+# 元数据操作，不跑模型回合：只调 thread/list、thread/read、
+# thread/name/set、thread/archive、thread/unarchive、
+# thread/compact/start、skills/list、app/list、
+# mcpServerStatus/list、config/mcpServer/reload、
+# mcpServer/oauth/login 这类只读或线程管理方法，不开 turn，
+# 因此不违反本文件开头“不直接跑模型回合”的原则。
+#
+# host.py CLI 契约（与正在加 thread-* 子命令的另一路对齐，见
+# 各 handler 的 argv 注释）：子命令只往 stdout 打一个 JSON
+# 对象；失败时 exit 非零且 stdout 是 {"error": ...}。本模块
+# 解析 whole-stdout JSON，解析失败或 host.py 缺失一律 502，
+# 绝不伪造成功。argv 里的子命令名与参数名是本批的约定，
+# host.py 侧实现时须与之一致（单测用 stub 脚本按此约定伪造）。
+def _host_py_path():
+    """host.py 路径：环境变量 CODEX_HOST_PY 优先，否则取
+    cloud/server.py 所在目录的上一级 host.py（与 worker 的
+    --host-py 缺省规则同款）。"""
+    env_path = os.environ.get("CODEX_HOST_PY")
+    if env_path:
+        return env_path
+    here = os.path.dirname(os.path.abspath(__file__))
+    return os.path.join(os.path.dirname(here), "host.py")
+
+
+def _run_host_cli(argv, codex_home, timeout=60):
+    """调 host.py CLI 子命令，返回 (ok, result_or_error)。
+
+    ok 为 True 时 result 是 stdout 解析出的 JSON 对象；ok 为
+    False 时 result 是可直接放进 502 error 字段的错误文本
+    （CLI 自带的 error 原样透传，解析失败/缺文件/超时则给
+    明确的服务端错误，绝不把 stderr 或原始 key 混进响应）。
+    """
+    host_py = _host_py_path()
+    if not host_py or not os.path.isfile(host_py):
+        return False, "host.py not found"
+    cmd = [sys.executable, host_py]
+    if codex_home:
+        cmd += ["--codex-home", codex_home]
+    cmd += list(argv)
+    try:
+        proc = subprocess.run(cmd, capture_output=True, text=True,
+                              timeout=timeout)
+    except subprocess.TimeoutExpired:
+        return False, "host.py timed out"
+    except OSError as exc:
+        return False, "cannot run host.py: %s" % exc
+    stdout = (proc.stdout or "").strip()
+    try:
+        parsed = json.loads(stdout) if stdout else None
+    except ValueError:
+        return False, "host.py output is not valid JSON"
+    if proc.returncode != 0:
+        err = None
+        if isinstance(parsed, dict):
+            err = parsed.get("error")
+        return False, str(err) if err else (
+            "host.py failed with exit %d" % proc.returncode)
+    if not isinstance(parsed, dict):
+        return False, "host.py output is not a JSON object"
+    if parsed.get("error"):
+        return False, str(parsed["error"])
+    return True, parsed
+
+
+def _tenant_safe_project_name(project_id):
+    """项目 id -> 租户目录安全名，与 worker._tenant_safe_name
+    同规则；映射不出安全名时返回 None。"""
+    if not project_id:
+        return None
+    safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(project_id)).strip(".")
+    if not safe or safe in (".", ".."):
+        return None
+    return safe
+
+
+def _tenant_codex_home(server, project_id):
+    """项目对应的租户 codex-home：tenants_root/<安全项目名>/
+    codex-home。线程持久化在 CODEX_HOME 里，租户按项目分
+    home，因此线程操作必须带这个 home，不能用共享 home。"""
+    safe = _tenant_safe_project_name(project_id)
+    if not safe:
+        return None
+    return os.path.join(_tenants_root_for(server), safe, "codex-home")
+
+
+def _tenant_work_dir(server, project_id):
+    """项目对应的租户工作目录：tenants_root/<安全项目名>/work，
+    skills/list 的 cwds 过滤用它（不传 cwd 时 app-server 会
+    默认用自身 cwd，那是错的 home 目录）。"""
+    safe = _tenant_safe_project_name(project_id)
+    if not safe:
+        return None
+    return os.path.join(_tenants_root_for(server), safe, "work")
+
+
+def _thread_worktree_for_task(server, task):
+    """本地任务的 worktree 推导：host.py dispatch 有仓库时在
+    <host.py 同级>/work/host-worktrees/<host 任务名> 建 worktree，
+    host 任务名 = 任务 id 的安全字符清洗（与 sandbox 的
+    _host_name 同规则）。无仓库任务的 thread cwd 是 worker 的
+    仓库路径或 host.py 当前目录，服务端表里没有、推不出来，
+    此时按此路径过滤会返回零个 thread，调用方拿到 404——
+    这是诚实结果，不是静默错指。部署可注入
+    server.host_worktrees_dir（/pr 路径同款属性）或环境变量
+    CODEX_HOST_WORKTREES_DIR 覆盖 worktree 根。"""
+    base = getattr(server, "host_worktrees_dir", None)
+    if not base:
+        base = os.environ.get("CODEX_HOST_WORKTREES_DIR")
+    if not base:
+        base = os.path.join(os.path.dirname(_host_py_path()),
+                            "work", "host-worktrees")
+    safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(task["id"]))
+    return os.path.join(base, safe)
+
+
+def _extract_thread_list(result):
+    """从 thread-list 的 CLI 结果里取 thread 列表：标准形态是
+    ThreadListResponse 的 data；threads 键与 result 包裹是
+    容忍形态（CLI 形状未定前不因键名不同就 502）。取不到
+    列表时返回 None（调用方按 502 处理）。"""
+    if not isinstance(result, dict):
+        return None
+    for key in ("data", "threads"):
+        value = result.get(key)
+        if isinstance(value, list):
+            return value
+    inner = result.get("result")
+    if isinstance(inner, dict):
+        for key in ("data", "threads"):
+            value = inner.get(key)
+            if isinstance(value, list):
+                return value
+    return None
+
+
+def _extract_thread(result):
+    """从 thread-read 的 CLI 结果里取 thread 对象：标准形态是
+    ThreadReadResponse 的 thread；结果本身就是 thread 对象
+    时也接受。取不到时返回 None。"""
+    if not isinstance(result, dict):
+        return None
+    thread = result.get("thread")
+    if isinstance(thread, dict):
+        return thread
+    if isinstance(result.get("id"), str):
+        return result
+    inner = result.get("result")
+    if isinstance(inner, dict):
+        thread = inner.get("thread")
+        if isinstance(thread, dict):
+            return thread
+        if isinstance(inner.get("id"), str):
+            return inner
+    return None
+
+
+def _pick_latest_thread(threads):
+    """cwd 过滤返回多个 thread 时的取用规则：取最新一个非
+    archived。协议说明：Thread 对象本身没有 archived 字段，
+    非归档由 thread/list 的 archived 过滤保证（不传 archived
+    时只回非归档 thread）；这里只防御性地跳过显式带
+    archived 真值的条目（CLI 形状若补了该字段时仍正确）。
+    最新 = updatedAt 最大，缺失时退 createdAt，再缺失时按
+    列表原顺序取后者（同秒时后到的更新）。"""
+    best = None
+    best_key = None
+    for idx, item in enumerate(threads):
+        if not isinstance(item, dict):
+            continue
+        if item.get("archived"):
+            continue
+        ts = item.get("updatedAt")
+        if ts is None:
+            ts = item.get("createdAt")
+        try:
+            ts = float(ts)
+        except (TypeError, ValueError):
+            ts = 0
+        key = (ts, idx)
+        if best_key is None or key > best_key:
+            best = item
+            best_key = key
+    return best
+
+
+def _extract_result_list(result, keys):
+    """从 skills/apps/mcp 列表类 CLI 结果里取列表：标准形态
+    是各 Response 的 data；传入的备用键是容忍形态。取不到
+    时返回 None。"""
+    if not isinstance(result, dict):
+        return None
+    for key in keys:
+        value = result.get(key)
+        if isinstance(value, list):
+            return value
+    inner = result.get("result")
+    if isinstance(inner, dict):
+        for key in keys:
+            value = inner.get(key)
+            if isinstance(value, list):
+                return value
+    return None
+
+
 def _read_trace_entries(path):
     """读 host 写的 jsonl 链路文件：跳过坏行；按 id 去重，后到的
     完成态覆盖先到的开始态，保首次顺序。文件不存在返回空列表。"""
@@ -1649,6 +1862,22 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 _task_part, _rev_part = rest.split("/reviews/", 1)
                 self._update_review_findings(
                     user_id, _task_part, _rev_part.split("/")[0])
+            elif rest.endswith("/threads") and method == "GET":
+                self._task_threads(user_id, rest[: -len("/threads")])
+            elif rest.endswith("/thread") and method == "GET":
+                self._task_thread(user_id, rest[: -len("/thread")], query)
+            elif rest.endswith("/thread/rename") and method == "POST":
+                self._task_thread_rename(
+                    user_id, rest[: -len("/thread/rename")])
+            elif rest.endswith("/thread/archive") and method == "POST":
+                self._task_thread_archive(
+                    user_id, rest[: -len("/thread/archive")])
+            elif rest.endswith("/thread/unarchive") and method == "POST":
+                self._task_thread_unarchive(
+                    user_id, rest[: -len("/thread/unarchive")])
+            elif rest.endswith("/thread/compact") and method == "POST":
+                self._task_thread_compact(
+                    user_id, rest[: -len("/thread/compact")])
             elif rest.endswith("/fork") and method == "POST":
                 self._fork_task(user_id, rest[: -len("/fork")])
             elif rest.endswith("/approvals") and method == "GET":
@@ -2438,6 +2667,69 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 return
             self._triage_inbox(project_id)
             return
+        # /projects/{id}/skills  GET（项目 read）
+        if len(parts) == 2 and parts[1] == "skills":
+            if method != "GET":
+                self._send_json(404, {"error": "not found"})
+                return
+            err = self._check_project(user_id, project_id, "read")
+            if err:
+                self._send_json(err[0], {"error": err[1]})
+                return
+            self._skills_list(project_id)
+            return
+        # /projects/{id}/apps  GET（项目 read）
+        if len(parts) == 2 and parts[1] == "apps":
+            if method != "GET":
+                self._send_json(404, {"error": "not found"})
+                return
+            err = self._check_project(user_id, project_id, "read")
+            if err:
+                self._send_json(err[0], {"error": err[1]})
+                return
+            self._apps_list(project_id)
+            return
+        # /projects/{id}/mcp-servers  GET（项目 read）
+        if len(parts) == 2 and parts[1] == "mcp-servers":
+            if method != "GET":
+                self._send_json(404, {"error": "not found"})
+                return
+            err = self._check_project(user_id, project_id, "read")
+            if err:
+                self._send_json(err[0], {"error": err[1]})
+                return
+            self._mcp_servers_list(project_id)
+            return
+        # /projects/{id}/mcp-servers/reload  POST（项目 manage_project）
+        if len(parts) == 3 and parts[1] == "mcp-servers" \
+                and parts[2] == "reload":
+            if method != "POST":
+                self._send_json(404, {"error": "not found"})
+                return
+            err = self._check_project(user_id, project_id,
+                                      "manage_project")
+            if err:
+                self._send_json(err[0], {"error": err[1]})
+                return
+            self._mcp_servers_reload(user_id, project_id)
+            return
+        # /projects/{id}/mcp-servers/{name}/oauth-login  POST
+        # （项目 manage_project；name 为空或穿越时 404）
+        if len(parts) == 4 and parts[1] == "mcp-servers" \
+                and parts[3] == "oauth-login":
+            if method != "POST":
+                self._send_json(404, {"error": "not found"})
+                return
+            if not parts[2] or parts[2] in (".", ".."):
+                self._send_json(404, {"error": "not found"})
+                return
+            err = self._check_project(user_id, project_id,
+                                      "manage_project")
+            if err:
+                self._send_json(err[0], {"error": err[1]})
+                return
+            self._mcp_server_oauth_login(user_id, project_id, parts[2])
+            return
         self._send_json(404, {"error": "not found"})
 
     def _memory_create(self, user_id, project_id):
@@ -3694,10 +3986,375 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                         "review_findings_updated",
                         "%s -> %s" % (task_id, review_id))
         self._send_json(200, updated)
+    # -- 任务线程操作（经 host.py CLI 透传）----------------------------
+    # 本批限制（P2 切片 C 接线时如实保留，见 cherry-pick 说明）：
+    # Docker/异机任务线程操作一律 409（见 _thread_task_guard）；
+    # 改操作仅 completed/failed/interrupted，merged/merge-failed
+    # 409（见 THREAD_MUTATION_STATUSES，控制台门控与此一致）；
+    # turn/steer 与 review/start 不在本批（review 端点是 P1 已有
+    # 功能，不由本批新增或改动）。
+    def _thread_task(self, user_id, task_id, action):
+        """取任务并验项目权限：返回 (task, None)，或 (None, err)，
+        err 为 (status, msg)，调用方直接 _send_json。任务不存在
+        404 在先（与 peer 消息同款顺序），权限不足 403。"""
+        row = self._fetch_task(task_id)
+        if row is None:
+            return None, (404, "task not found")
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], action):
+                return None, (403, "forbidden")
+        return task, None
+
+    def _thread_task_guard(self, task):
+        """线程操作的执行器门：只做 local 执行器任务。Docker
+        任务的 thread 在容器内、cloud 任务的 thread 不在本机，
+        本机租户 codex-home 里没有，查了也是错指，一律 409
+        明说，不回空结果假装无 thread。executor 为空（建任务
+        时未指定）时放行：老任务多为 NULL 且实际由 local
+        worker 认领，卡死它们等于端点对老任务不可用；若实际
+        是 Docker 跑的，thread/list 会返回零个而得 404。
+        worktree 在别机时同样 409（与链路的跨机诚实降级同款，
+        但改操作绝不能在错的主机 home 上执行，所以不用 200
+        空结果）。返回 None 表示通过，否则为 (status, msg)。"""
+        executor = task.get("executor")
+        if executor == "docker":
+            return (409, "thread ops are local-executor only;"
+                    " docker task threads live inside the container")
+        if executor == "cloud":
+            return (409, "thread ops are local-executor only;"
+                    " cloud task threads are not on this host")
+        worktree_host = task.get("worktree_host")
+        if worktree_host and worktree_host != socket.gethostname():
+            return (409, "task worktree is on host %s;"
+                    " thread ops must run there" % worktree_host)
+        return None
+
+    def _list_task_threads(self, task):
+        """对任务调 thread-list：返回 (threads, codex_home, cwd,
+        err)，err 为 (status, msg) 或 None。cwd 过滤用任务
+        worktree（见 _thread_worktree_for_task）；codex-home
+        用租户 home（见 _tenant_codex_home）。"""
+        codex_home = _tenant_codex_home(self.server,
+                                        task["project_id"])
+        if codex_home is None:
+            return None, None, None, (
+                502, "cannot resolve tenant codex-home")
+        cwd = _thread_worktree_for_task(self.server, task)
+        ok, result = _run_host_cli(["thread-list", "--cwd", cwd],
+                                   codex_home)
+        if not ok:
+            return None, None, None, (502, result)
+        threads = _extract_thread_list(result)
+        if threads is None:
+            return None, None, None, (
+                502, "host.py thread-list output has no thread list")
+        return threads, codex_home, cwd, None
+
+    def _resolve_task_thread(self, task):
+        """单线程端点的解析：thread-list 后按 _pick_latest_thread
+        取最新一个非 archived。返回 (thread, codex_home, cwd,
+        err)；零个 thread 时 err 为 (404, "thread not found")。"""
+        threads, codex_home, cwd, err = self._list_task_threads(task)
+        if err is not None:
+            return None, None, None, err
+        thread = _pick_latest_thread(threads)
+        if thread is None:
+            return None, None, None, (404, "thread not found")
+        return thread, codex_home, cwd, None
+
+    def _task_threads(self, user_id, task_id):
+        """GET /tasks/{id}/threads：thread/list 原样返回列表。
+        读权限用项目 read；零个 thread 时返回空列表（原样），
+        不像单线程端点那样 404——列表端点没有“解析不到”的
+        概念，空就是空。"""
+        task, err = self._thread_task(user_id, task_id, "read")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        guard = self._thread_task_guard(task)
+        if guard is not None:
+            self._send_json(guard[0], {"error": guard[1]})
+            return
+        threads, _home, cwd, err = self._list_task_threads(task)
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        self._send_json(200, {"task_id": task_id, "cwd": cwd,
+                              "threads": threads})
+
+    def _task_thread(self, user_id, task_id, query):
+        """GET /tasks/{id}/thread：先 thread/list 解析出 thread，
+        再 thread/read。默认只要元数据（includeTurns 不传）；
+        ?include_turns=1 时才传 includeTurns。读操作不限制
+        任务状态；执行错误原样透传为 502。"""
+        task, err = self._thread_task(user_id, task_id, "read")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        guard = self._thread_task_guard(task)
+        if guard is not None:
+            self._send_json(guard[0], {"error": guard[1]})
+            return
+        thread, codex_home, cwd, err = self._resolve_task_thread(task)
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        argv = ["thread-read", "--thread-id", thread["id"]]
+        include_turns = (query.get("include_turns") or ["0"])[0]
+        if include_turns == "1":
+            argv.append("--include-turns")
+        ok, result = _run_host_cli(argv, codex_home)
+        if not ok:
+            self._send_json(502, {"error": result})
+            return
+        full = _extract_thread(result)
+        if full is None:
+            self._send_json(502, {
+                "error": "host.py thread-read output has no thread"})
+            return
+        self._send_json(200, {"task_id": task_id, "cwd": cwd,
+                              "thread_id": thread["id"],
+                              "thread": full})
+
+    def _task_thread_mutate(self, user_id, task_id, argv, audit_action,
+                            audit_detail, timeout=60):
+        """线程改操作的公共流程：权限用项目 create_task；只允许
+        终态任务（completed/failed/interrupted），running/queued
+        一律 409；解析 thread 后调对应 CLI，成功记审计。"""
+        task, err = self._thread_task(user_id, task_id, "create_task")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        guard = self._thread_task_guard(task)
+        if guard is not None:
+            self._send_json(guard[0], {"error": guard[1]})
+            return
+        if task["status"] not in THREAD_MUTATION_STATUSES:
+            self._send_json(409, {
+                "error": "thread ops require a terminal task",
+                "status": task["status"]})
+            return
+        thread, codex_home, _cwd, err = self._resolve_task_thread(task)
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        ok, result = _run_host_cli(
+            list(argv) + ["--thread-id", thread["id"]],
+            codex_home, timeout=timeout)
+        if not ok:
+            self._send_json(502, {"error": result})
+            return
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, task["project_id"], audit_action,
+                        audit_detail % thread["id"])
+        self._send_json(200, {"task_id": task_id,
+                              "thread_id": thread["id"],
+                              "result": result})
+
+    def _task_thread_rename(self, user_id, task_id):
+        """POST /tasks/{id}/thread/rename {name}：thread/name/set。
+        name 必填、非空、200 字内；记 thread_renamed 审计。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        name = body.get("name")
+        if not name or not isinstance(name, str) or not name.strip():
+            self._send_json(400, {"error": "name required"})
+            return
+        name = name.strip()
+        if len(name) > 200:
+            self._send_json(400, {"error": "name too long"})
+            return
+        self._task_thread_mutate(
+            user_id, task_id, ["thread-rename", "--name", name],
+            "thread_renamed", "%s renamed to " + name)
+
+    def _task_thread_archive(self, user_id, task_id):
+        """POST /tasks/{id}/thread/archive：thread/archive。
+        记 thread_archived 审计。"""
+        self._task_thread_mutate(user_id, task_id, ["thread-archive"],
+                                 "thread_archived", "%s archived")
+
+    def _task_thread_unarchive(self, user_id, task_id):
+        """POST /tasks/{id}/thread/unarchive：thread/unarchive。
+        记 thread_unarchived 审计。"""
+        self._task_thread_mutate(user_id, task_id, ["thread-unarchive"],
+                                 "thread_unarchived", "%s unarchived")
+
+    def _task_thread_compact(self, user_id, task_id):
+        """POST /tasks/{id}/thread/compact：thread/compact/start。
+        压缩可能耗时，CLI 超时放宽到 120 秒；记 thread_compacted
+        审计。注意：这是线程压缩，不是开新回合，不跑模型
+        轮次之外的 turn（压缩本身由 app-server 内部完成）。"""
+        self._task_thread_mutate(user_id, task_id, ["thread-compact"],
+                                 "thread_compacted", "%s compacted",
+                                 timeout=120)
+
+    # -- 项目 skills / apps / MCP（经 host.py CLI 透传）--------------
+    # codex-home 按项目分租户目录，所以这批端点是项目 scope：
+    # 读用项目 read，改用项目 manage_project（owner）。
+    def _project_host_home(self, project_id):
+        """项目端点的租户 home 与工作目录：返回 (codex_home,
+        work_dir)，任一解析不出时返回 (None, None)。注意：这里
+        反映的是宿主租户 home，容器内 home 可能不同，skills/MCP
+        面看到的是宿主侧视图，不代表容器内视图。"""
+        codex_home = _tenant_codex_home(self.server, project_id)
+        work_dir = _tenant_work_dir(self.server, project_id)
+        if codex_home is None or work_dir is None:
+            return None, None
+        return codex_home, work_dir
+
+    def _skills_list(self, project_id):
+        """GET /projects/{id}/skills：skills/list 原样返回。"""
+        codex_home, work_dir = self._project_host_home(project_id)
+        if codex_home is None:
+            self._send_json(502, {
+                "error": "cannot resolve tenant codex-home"})
+            return
+        ok, result = _run_host_cli(["skills-list", "--cwd", work_dir],
+                                   codex_home)
+        if not ok:
+            self._send_json(502, {"error": result})
+            return
+        skills = _extract_result_list(result, ("data", "skills"))
+        if skills is None:
+            self._send_json(502, {
+                "error": "host.py skills-list output has no list"})
+            return
+        self._send_json(200, {"project_id": project_id,
+                              "skills": skills})
+
+    def _apps_list(self, project_id):
+        """GET /projects/{id}/apps：app/list 原样返回（schema 里
+        是 AppsListParams/AppsListResponse；项目 scope 下不带
+        threadId，不做按线程的特性门控）。"""
+        codex_home, _work = self._project_host_home(project_id)
+        if codex_home is None:
+            self._send_json(502, {
+                "error": "cannot resolve tenant codex-home"})
+            return
+        ok, result = _run_host_cli(["apps-list"], codex_home)
+        if not ok:
+            self._send_json(502, {"error": result})
+            return
+        apps = _extract_result_list(result, ("data", "apps"))
+        if apps is None:
+            self._send_json(502, {
+                "error": "host.py apps-list output has no list"})
+            return
+        body = {"project_id": project_id, "apps": apps}
+        cursor = result.get("nextCursor")
+        if cursor is not None:
+            body["nextCursor"] = cursor
+        self._send_json(200, body)
+
+    def _mcp_servers_list(self, project_id):
+        """GET /projects/{id}/mcp-servers：mcpServerStatus/list
+        原样返回（schema 里是 ListMcpServerStatusParams/
+        Response；不带 threadId，走配置发现）。"""
+        codex_home, _work = self._project_host_home(project_id)
+        if codex_home is None:
+            self._send_json(502, {
+                "error": "cannot resolve tenant codex-home"})
+            return
+        ok, result = _run_host_cli(["mcp-servers-list"], codex_home)
+        if not ok:
+            self._send_json(502, {"error": result})
+            return
+        servers = _extract_result_list(result, ("data", "servers"))
+        if servers is None:
+            self._send_json(502, {
+                "error": "host.py mcp-servers-list output has no list"})
+            return
+        body = {"project_id": project_id, "servers": servers}
+        cursor = result.get("nextCursor")
+        if cursor is not None:
+            body["nextCursor"] = cursor
+        self._send_json(200, body)
+
+    def _mcp_servers_reload(self, user_id, project_id):
+        """POST /projects/{id}/mcp-servers/reload：config/mcpServer/
+        reload。核对结论：该方法在协议 schema 里真实存在——
+        v1 与 v2 的 ClientRequest 方法枚举里都有它，params
+        为 null，没有专用的响应定义（结果通常为空），所以
+        正常调、结果原样透传；只有 CLI 本身报错时才 502，
+        绝不伪造“不支持”也不伪造成功。记审计。"""
+        codex_home, _work = self._project_host_home(project_id)
+        if codex_home is None:
+            self._send_json(502, {
+                "error": "cannot resolve tenant codex-home"})
+            return
+        ok, result = _run_host_cli(["mcp-servers-reload"], codex_home)
+        if not ok:
+            self._send_json(502, {"error": result})
+            return
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, project_id, "mcp_servers_reloaded",
+                        "mcp servers reloaded")
+        self._send_json(200, {"project_id": project_id,
+                              "reloaded": True, "result": result})
+
+    def _mcp_server_oauth_login(self, user_id, project_id, name):
+        """POST /projects/{id}/mcp-servers/{name}/oauth-login：
+        mcpServer/oauth/login。协议返回（McpServerOauthLoginResponse：
+        authorizationUrl 必填、loginId 可空）原样透传给调用方，
+        浏览器里完成登录的 URL/步骤由调用方展示。完成态轮询
+        不做：协议里完成信号是 McpServerOauthLoginCompletedNotification
+        （服务端通知，不是请求/响应轮询），本端点只做到“发起
+        登录并返回授权 URL”这一步，绝不假装登录已完成。
+        记审计时只记服务器名与 loginId，不记授权 URL。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        argv = ["mcp-server-oauth-login", "--server-name", name]
+        scopes = body.get("scopes")
+        if scopes is not None:
+            if not isinstance(scopes, list) or any(
+                    not isinstance(s, str) for s in scopes):
+                self._send_json(400, {
+                    "error": "scopes must be a list of strings"})
+                return
+            if scopes:
+                argv += ["--scopes", ",".join(scopes)]
+        timeout_secs = body.get("timeoutSecs")
+        if timeout_secs is not None:
+            if not isinstance(timeout_secs, int) or timeout_secs <= 0:
+                self._send_json(400, {
+                    "error": "timeoutSecs must be a positive integer"})
+                return
+            argv += ["--timeout-secs", str(timeout_secs)]
+        codex_home, _work = self._project_host_home(project_id)
+        if codex_home is None:
+            self._send_json(502, {
+                "error": "cannot resolve tenant codex-home"})
+            return
+        ok, result = _run_host_cli(argv, codex_home, timeout=120)
+        if not ok:
+            self._send_json(502, {"error": result})
+            return
+        if not isinstance(result.get("authorizationUrl"), str):
+            self._send_json(502, {
+                "error": "host.py oauth-login output has no"
+                         " authorizationUrl"})
+            return
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, project_id,
+                        "mcp_server_oauth_login_started",
+                        "%s loginId=%s" % (
+                            name, result.get("loginId") or ""))
+        response = dict(result)
+        response["project_id"] = project_id
+        response["server_name"] = name
+        self._send_json(200, response)
 
     def _fork_task(self, user_id, task_id):
         """会话分支：从已结束任务的 thread 分叉出新任务。
 
+
         新任务继承源任务的 executor/模型/强度/权限/计划模式，
         不继承 goal（分支是普通任务，首轮 prompt 就是这里传的
         message）。worker 认领后先调 host.py fork（分叉 thread

cloud/tests/test_console.py

@@ -65,7 +65,15 @@ class ConsoleTestCase(unittest.TestCase):
         # 壳必须引用它实际调的接口，防 JS 与服务端路由漂移。
         for marker in ("/health", "/projects", "/tasks", "/workers",
                        "X-API-Key", "agentBox", "/agents",
-                       "peerMessage", "/messages"):
+                       "peerMessage", "/messages",
+                       "threadBox", "threadGateNote", "threadList",
+                       "THREAD_MUTATION_STATUSES",
+                       "/thread", "/threads",
+                       "/thread/rename", "/thread/archive",
+                       "/thread/unarchive", "/thread/compact",
+                       "skillList", "appList", "mcpList", "mcpOauthBox",
+                       "/skills", "/apps", "/mcp-servers",
+                       "/mcp-servers/reload", "/oauth-login"):
             self.assertIn(marker, text)
 
     def test_console_trailing_slash_same_page(self):

cloud/tests/test_skills_mcp.py

@@ -0,0 +1,337 @@
+"""Skills/apps/MCP server tests: /projects/<id>/skills, /apps,
+/mcp-servers, /mcp-servers/reload, /mcp-servers/<name>/oauth-login.
+
+No real host.py CLI, no real codex binary: a stub script written into
+tmp pretends to be host.py and prints canned JSON per the frozen CLI
+contract - one JSON object on stdout, nonzero exit plus {"error": ...}
+on failure. The stub reads a scenario JSON file and appends every
+invocation to a log file so tests can assert the argv the server
+passed, including the tenant work dir used as skills/list cwd.
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+
+STUB = r'''import json
+import os
+import sys
+
+def main():
+    argv = sys.argv[1:]
+    cmd = None
+    rest = []
+    i = 0
+    while i < len(argv):
+        a = argv[i]
+        if a == "--codex-home":
+            i += 2
+            continue
+        if a.startswith("--codex-home="):
+            i += 1
+            continue
+        if cmd is None:
+            cmd = a
+            i += 1
+            continue
+        rest.append(a)
+        i += 1
+    log = os.environ.get("THREADOPS_STUB_LOG")
+    if log:
+        with open(log, "a", encoding="utf-8") as f:
+            f.write(json.dumps({"cmd": cmd, "argv": rest}) + "\n")
+    scenario = {}
+    path = os.environ.get("THREADOPS_STUB_SCENARIO")
+    if path and os.path.exists(path):
+        with open(path, "r", encoding="utf-8") as f:
+            scenario = json.load(f)
+    if scenario.get("fail_cmd") == cmd:
+        print(json.dumps({"error": scenario.get("fail_error", "boom")}))
+        sys.exit(1)
+    if cmd == "skills-list":
+        print(json.dumps({"data": scenario.get("skills", [
+            {"cwd": "/work", "skills": [{"name": "demo-skill"}]},
+        ])}))
+    elif cmd == "apps-list":
+        print(json.dumps({"data": [{"id": "app-1", "name": "Demo App"}],
+                          "nextCursor": "cur-1"}))
+    elif cmd == "mcp-servers-list":
+        print(json.dumps({"data": [{"name": "srv-1",
+                                    "runtimeStatus": "running"}]}))
+    elif cmd == "mcp-servers-reload":
+        print(json.dumps({}))
+    elif cmd == "mcp-server-oauth-login":
+        if scenario.get("oauth_no_url"):
+            print(json.dumps({}))
+        else:
+            print(json.dumps({
+                "authorizationUrl": "https://example.invalid/oauth/authorize?x=1",
+                "loginId": "login-123"}))
+    else:
+        print(json.dumps({"error": "unknown subcommand"}))
+        sys.exit(1)
+
+main()
+'''
+
+
+class SkillsMcpTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.reader_id, self.reader_key = self.auth.create_user("reader")
+        self.outsider_id, self.outsider_key = self.auth.create_user("outsider")
+        self.stub_path = os.path.join(self.tmp.name, "stub_host.py")
+        with open(self.stub_path, "w", encoding="utf-8") as f:
+            f.write(STUB)
+        self.scenario_path = os.path.join(self.tmp.name, "scenario.json")
+        self.log_path = os.path.join(self.tmp.name, "stub.log")
+        self._write_scenario({})
+        self._old_host_py = os.environ.get("CODEX_HOST_PY")
+        os.environ["CODEX_HOST_PY"] = self.stub_path
+        os.environ["THREADOPS_STUB_SCENARIO"] = self.scenario_path
+        os.environ["THREADOPS_STUB_LOG"] = self.log_path
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.server.tenants_root = os.path.join(self.tmp.name, "tenants")
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201, self.project)
+        self.auth.set_member(self.project["id"], self.owner_id,
+                             self.reader_id, "viewer")
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        if self._old_host_py is None:
+            os.environ.pop("CODEX_HOST_PY", None)
+        else:
+            os.environ["CODEX_HOST_PY"] = self._old_host_py
+        os.environ.pop("THREADOPS_STUB_SCENARIO", None)
+        os.environ.pop("THREADOPS_STUB_LOG", None)
+        self.tmp.cleanup()
+
+    def _write_scenario(self, scenario):
+        with open(self.scenario_path, "w", encoding="utf-8") as f:
+            json.dump(scenario, f)
+
+    def _stub_calls(self, cmd=None):
+        if not os.path.exists(self.log_path):
+            return []
+        calls = []
+        with open(self.log_path, "r", encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if line:
+                    calls.append(json.loads(line))
+        if cmd is not None:
+            calls = [c for c in calls if c["cmd"] == cmd]
+        return calls
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _audit_actions(self, project_id):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            rows = conn.execute(
+                "SELECT action, detail FROM audit_log WHERE project_id = ?",
+                (project_id,)).fetchall()
+        finally:
+            conn.close()
+        return [(r[0], r[1]) for r in rows]
+
+    def _pid(self):
+        return self.project["id"]
+
+    def test_unauthorized_without_key(self):
+        status, body = self._req(
+            "GET", "/projects/%s/skills" % self._pid(), None, None)
+        self.assertEqual(status, 401, body)
+
+    def test_forbidden_outsider_reads(self):
+        for suffix in ("skills", "apps", "mcp-servers"):
+            status, body = self._req(
+                "GET", "/projects/%s/%s" % (self._pid(), suffix),
+                None, self.outsider_key)
+            self.assertEqual(status, 403, body)
+
+    def test_project_not_found(self):
+        status, body = self._req(
+            "GET", "/projects/nope/skills", None, self.owner_key)
+        self.assertEqual(status, 404, body)
+
+    def test_skills_list_passthrough_and_cwd(self):
+        status, body = self._req(
+            "GET", "/projects/%s/skills" % self._pid(), None,
+            self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["project_id"], self._pid())
+        self.assertEqual(body["skills"], [
+            {"cwd": "/work", "skills": [{"name": "demo-skill"}]}])
+        calls = self._stub_calls("skills-list")
+        self.assertEqual(len(calls), 1)
+        self.assertIn("--cwd", calls[0]["argv"])
+        idx = calls[0]["argv"].index("--cwd")
+        expected = os.path.join(
+            self.tmp.name, "tenants", self._pid(), "work")
+        self.assertEqual(calls[0]["argv"][idx + 1], expected)
+
+    def test_reader_can_read_skills(self):
+        status, body = self._req(
+            "GET", "/projects/%s/skills" % self._pid(), None,
+            self.reader_key)
+        self.assertEqual(status, 200, body)
+
+    def test_apps_list_passthrough(self):
+        status, body = self._req(
+            "GET", "/projects/%s/apps" % self._pid(), None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["apps"], [{"id": "app-1", "name": "Demo App"}])
+        self.assertEqual(body["nextCursor"], "cur-1")
+
+    def test_mcp_servers_list_passthrough(self):
+        status, body = self._req(
+            "GET", "/projects/%s/mcp-servers" % self._pid(), None,
+            self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["servers"], [
+            {"name": "srv-1", "runtimeStatus": "running"}])
+
+    def test_reload_ok_and_audit(self):
+        status, body = self._req(
+            "POST", "/projects/%s/mcp-servers/reload" % self._pid(),
+            {}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertTrue(body["reloaded"])
+        self.assertEqual(body["result"], {})
+        actions = [a for a, _d in self._audit_actions(self._pid())]
+        self.assertIn("mcp_servers_reloaded", actions)
+
+    def test_reload_forbidden_for_reader(self):
+        status, body = self._req(
+            "POST", "/projects/%s/mcp-servers/reload" % self._pid(),
+            {}, self.reader_key)
+        self.assertEqual(status, 403, body)
+
+    def test_reload_cli_failure_502_not_fake_success(self):
+        self._write_scenario(
+            {"fail_cmd": "mcp-servers-reload",
+             "fail_error": "reload rejected"})
+        status, body = self._req(
+            "POST", "/projects/%s/mcp-servers/reload" % self._pid(),
+            {}, self.owner_key)
+        self.assertEqual(status, 502, body)
+        self.assertEqual(body["error"], "reload rejected")
+        self.assertNotIn("reloaded", body)
+
+    def test_oauth_login_passthrough(self):
+        status, body = self._req(
+            "POST",
+            "/projects/%s/mcp-servers/srv-1/oauth-login" % self._pid(),
+            {"scopes": ["read", "write"]}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(
+            body["authorizationUrl"],
+            "https://example.invalid/oauth/authorize?x=1")
+        self.assertEqual(body["loginId"], "login-123")
+        self.assertEqual(body["project_id"], self._pid())
+        self.assertEqual(body["server_name"], "srv-1")
+        calls = self._stub_calls("mcp-server-oauth-login")
+        self.assertEqual(len(calls), 1)
+        self.assertIn("--server-name", calls[0]["argv"])
+        idx = calls[0]["argv"].index("--server-name")
+        self.assertEqual(calls[0]["argv"][idx + 1], "srv-1")
+        self.assertIn("--scopes", calls[0]["argv"])
+        idx = calls[0]["argv"].index("--scopes")
+        self.assertEqual(calls[0]["argv"][idx + 1], "read,write")
+        actions = self._audit_actions(self._pid())
+        self.assertTrue(any(
+            a == "mcp_server_oauth_login_started" and "srv-1" in d
+            for a, d in actions))
+
+    def test_oauth_login_forbidden_for_reader(self):
+        status, body = self._req(
+            "POST",
+            "/projects/%s/mcp-servers/srv-1/oauth-login" % self._pid(),
+            {}, self.reader_key)
+        self.assertEqual(status, 403, body)
+
+    def test_oauth_login_validation(self):
+        status, body = self._req(
+            "POST",
+            "/projects/%s/mcp-servers/srv-1/oauth-login" % self._pid(),
+            {"scopes": "not-a-list"}, self.owner_key)
+        self.assertEqual(status, 400, body)
+        status, body = self._req(
+            "POST",
+            "/projects/%s/mcp-servers/srv-1/oauth-login" % self._pid(),
+            {"timeoutSecs": 0}, self.owner_key)
+        self.assertEqual(status, 400, body)
+
+    def test_oauth_login_cli_failure_502(self):
+        self._write_scenario(
+            {"fail_cmd": "mcp-server-oauth-login",
+             "fail_error": "oauth rejected"})
+        status, body = self._req(
+            "POST",
+            "/projects/%s/mcp-servers/srv-1/oauth-login" % self._pid(),
+            {}, self.owner_key)
+        self.assertEqual(status, 502, body)
+        self.assertEqual(body["error"], "oauth rejected")
+
+    def test_oauth_login_missing_url_502(self):
+        self._write_scenario({"oauth_no_url": True})
+        status, body = self._req(
+            "POST",
+            "/projects/%s/mcp-servers/srv-1/oauth-login" % self._pid(),
+            {}, self.owner_key)
+        self.assertEqual(status, 502, body)
+        self.assertNotIn("authorizationUrl", body)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_thread_ops.py

@@ -0,0 +1,430 @@
+"""Thread ops server tests: GET/POST /tasks/<id>/thread* endpoints.
+
+No real host.py CLI, no real codex binary: a stub script written into
+tmp pretends to be host.py and prints canned JSON per the frozen CLI
+contract - one JSON object on stdout, nonzero exit plus {"error": ...}
+on failure. The stub reads a scenario JSON file and appends every
+invocation to a log file so tests can assert the argv the server
+passed, including the derived --cwd worktree path.
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+
+STUB = r'''import json
+import os
+import sys
+
+def main():
+    argv = sys.argv[1:]
+    cmd = None
+    rest = []
+    i = 0
+    while i < len(argv):
+        a = argv[i]
+        if a == "--codex-home":
+            i += 2
+            continue
+        if a.startswith("--codex-home="):
+            i += 1
+            continue
+        if cmd is None:
+            cmd = a
+            i += 1
+            continue
+        rest.append(a)
+        i += 1
+    log = os.environ.get("THREADOPS_STUB_LOG")
+    if log:
+        with open(log, "a", encoding="utf-8") as f:
+            f.write(json.dumps({"cmd": cmd, "argv": rest}) + "\n")
+    scenario = {}
+    path = os.environ.get("THREADOPS_STUB_SCENARIO")
+    if path and os.path.exists(path):
+        with open(path, "r", encoding="utf-8") as f:
+            scenario = json.load(f)
+    if scenario.get("fail_cmd") == cmd:
+        print(json.dumps({"error": scenario.get("fail_error", "boom")}))
+        sys.exit(1)
+    if scenario.get("raw_stdout_for") == cmd:
+        print(scenario.get("raw_stdout", ""))
+        return
+    opts = {}
+    j = 0
+    while j < len(rest):
+        a = rest[j]
+        if a.startswith("--") and j + 1 < len(rest) and not rest[j + 1].startswith("--"):
+            opts[a] = rest[j + 1]
+            j += 2
+        elif a.startswith("--"):
+            opts[a] = True
+            j += 1
+        else:
+            j += 1
+    if cmd == "thread-list":
+        print(json.dumps({"data": scenario.get("threads", [
+            {"id": "t-old", "updatedAt": 100, "createdAt": 90},
+            {"id": "t-new", "updatedAt": 200, "createdAt": 190},
+        ])}))
+    elif cmd == "thread-read":
+        thread = dict(scenario.get("thread", {"id": "t-new", "updatedAt": 200}))
+        if opts.get("--include-turns"):
+            thread["turns"] = [{"id": "turn-1"}]
+        print(json.dumps({"thread": thread}))
+    elif cmd == "thread-rename":
+        print(json.dumps({"ok": True, "name": opts.get("--name")}))
+    elif cmd in ("thread-archive", "thread-unarchive", "thread-compact"):
+        print(json.dumps({"ok": True}))
+    else:
+        print(json.dumps({"error": "unknown subcommand"}))
+        sys.exit(1)
+
+main()
+'''
+
+
+class ThreadOpsTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.reader_id, self.reader_key = self.auth.create_user("reader")
+        self.outsider_id, self.outsider_key = self.auth.create_user("outsider")
+        self.stub_path = os.path.join(self.tmp.name, "stub_host.py")
+        with open(self.stub_path, "w", encoding="utf-8") as f:
+            f.write(STUB)
+        self.scenario_path = os.path.join(self.tmp.name, "scenario.json")
+        self.log_path = os.path.join(self.tmp.name, "stub.log")
+        self._write_scenario({})
+        self._old_host_py = os.environ.get("CODEX_HOST_PY")
+        os.environ["CODEX_HOST_PY"] = self.stub_path
+        os.environ["THREADOPS_STUB_SCENARIO"] = self.scenario_path
+        os.environ["THREADOPS_STUB_LOG"] = self.log_path
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.server.tenants_root = os.path.join(self.tmp.name, "tenants")
+        self.server.host_worktrees_dir = os.path.join(
+            self.tmp.name, "worktrees")
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201, self.project)
+        self.auth.set_member(self.project["id"], self.owner_id,
+                             self.reader_id, "viewer")
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        if self._old_host_py is None:
+            os.environ.pop("CODEX_HOST_PY", None)
+        else:
+            os.environ["CODEX_HOST_PY"] = self._old_host_py
+        os.environ.pop("THREADOPS_STUB_SCENARIO", None)
+        os.environ.pop("THREADOPS_STUB_LOG", None)
+        self.tmp.cleanup()
+
+    def _write_scenario(self, scenario):
+        with open(self.scenario_path, "w", encoding="utf-8") as f:
+            json.dump(scenario, f)
+
+    def _stub_calls(self, cmd=None):
+        if not os.path.exists(self.log_path):
+            return []
+        calls = []
+        with open(self.log_path, "r", encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if line:
+                    calls.append(json.loads(line))
+        if cmd is not None:
+            calls = [c for c in calls if c["cmd"] == cmd]
+        return calls
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _create_task(self, **extra):
+        body = {"project_id": self.project["id"], "prompt": "do work"}
+        body.update(extra)
+        status, task = self._req("POST", "/tasks", body, self.owner_key)
+        self.assertEqual(status, 201, task)
+        return task
+
+    def _set_status(self, task_id, status):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
+                         (status, task_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _audit_actions(self, project_id):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            rows = conn.execute(
+                "SELECT action FROM audit_log WHERE project_id = ?",
+                (project_id,)).fetchall()
+        finally:
+            conn.close()
+        return [r[0] for r in rows]
+
+    def test_unauthorized_without_key(self):
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/thread" % task["id"], None, None)
+        self.assertEqual(status, 401, body)
+
+    def test_forbidden_outsider_read(self):
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/threads" % task["id"], None,
+            self.outsider_key)
+        self.assertEqual(status, 403, body)
+
+    def test_forbidden_reader_cannot_mutate(self):
+        task = self._create_task(executor="local")
+        self._set_status(task["id"], "completed")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/archive" % task["id"], {},
+            self.reader_key)
+        self.assertEqual(status, 403, body)
+
+    def test_forbidden_outsider_cannot_mutate(self):
+        # P2 接线新增：非成员改操作同样 403，且不留审计。
+        task = self._create_task(executor="local")
+        self._set_status(task["id"], "completed")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/archive" % task["id"], {},
+            self.outsider_key)
+        self.assertEqual(status, 403, body)
+        self.assertNotIn("thread_archived",
+                         self._audit_actions(self.project["id"]))
+
+    def test_task_not_found(self):
+        status, body = self._req(
+            "GET", "/tasks/nope/thread", None, self.owner_key)
+        self.assertEqual(status, 404, body)
+        status, body = self._req(
+            "POST", "/tasks/nope/thread/archive", {}, self.owner_key)
+        self.assertEqual(status, 404, body)
+
+    def test_docker_task_409(self):
+        task = self._create_task(executor="docker")
+        self._set_status(task["id"], "completed")
+        status, body = self._req(
+            "GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 409, body)
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/archive" % task["id"], {},
+            self.owner_key)
+        self.assertEqual(status, 409, body)
+        self.assertEqual(self._stub_calls(), [])
+
+    def test_terminal_gate_409_and_read_allowed(self):
+        task = self._create_task(executor="local")
+        self._set_status(task["id"], "running")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/archive" % task["id"], {},
+            self.owner_key)
+        self.assertEqual(status, 409, body)
+        self.assertEqual(body["status"], "running")
+        status, body = self._req(
+            "GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+
+    def test_merged_task_mutation_409_but_read_ok(self):
+        task = self._create_task(executor="local")
+        self._set_status(task["id"], "merged")
+        status, _body = self._req(
+            "POST", "/tasks/%s/thread/archive" % task["id"], {},
+            self.owner_key)
+        self.assertEqual(status, 409)
+        status, body = self._req(
+            "GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+
+    def test_threads_list_passthrough_and_cwd(self):
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(
+            [t["id"] for t in body["threads"]], ["t-old", "t-new"])
+        expected_cwd = os.path.join(
+            self.tmp.name, "worktrees", task["id"])
+        self.assertEqual(body["cwd"], expected_cwd)
+        calls = self._stub_calls("thread-list")
+        self.assertEqual(len(calls), 1)
+        self.assertIn("--cwd", calls[0]["argv"])
+        idx = calls[0]["argv"].index("--cwd")
+        self.assertEqual(calls[0]["argv"][idx + 1], expected_cwd)
+
+    def test_thread_latest_non_archived_rule(self):
+        self._write_scenario({"threads": [
+            {"id": "t-old", "updatedAt": 100, "createdAt": 90},
+            {"id": "t-arch", "updatedAt": 300, "createdAt": 290,
+             "archived": True},
+            {"id": "t-new", "updatedAt": 200, "createdAt": 190},
+        ]})
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["thread_id"], "t-new")
+
+    def test_thread_tie_break_later_in_list_wins(self):
+        self._write_scenario({"threads": [
+            {"id": "t-a", "updatedAt": 50, "createdAt": 50},
+            {"id": "t-b", "updatedAt": 50, "createdAt": 50},
+        ]})
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["thread_id"], "t-b")
+
+    def test_thread_not_found_when_empty(self):
+        self._write_scenario({"threads": []})
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 404, body)
+        status, body = self._req(
+            "GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["threads"], [])
+
+    def test_thread_read_include_turns(self):
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertNotIn("turns", body["thread"])
+        status, body = self._req(
+            "GET", "/tasks/%s/thread?include_turns=1" % task["id"],
+            None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["thread"]["turns"], [{"id": "turn-1"}])
+        calls = self._stub_calls("thread-read")
+        self.assertIn("--include-turns", calls[1]["argv"])
+        self.assertNotIn("--include-turns", calls[0]["argv"])
+
+    def test_rename_ok_and_audit(self):
+        task = self._create_task(executor="local")
+        self._set_status(task["id"], "completed")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/rename" % task["id"],
+            {"name": "new name"}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["result"]["name"], "new name")
+        self.assertIn("thread_renamed",
+                      self._audit_actions(self.project["id"]))
+        calls = self._stub_calls("thread-rename")
+        self.assertEqual(len(calls), 1)
+        self.assertIn("new name", calls[0]["argv"])
+
+    def test_rename_validation(self):
+        task = self._create_task(executor="local")
+        self._set_status(task["id"], "completed")
+        status, _body = self._req(
+            "POST", "/tasks/%s/thread/rename" % task["id"],
+            {}, self.owner_key)
+        self.assertEqual(status, 400)
+        status, _body = self._req(
+            "POST", "/tasks/%s/thread/rename" % task["id"],
+            {"name": "   "}, self.owner_key)
+        self.assertEqual(status, 400)
+
+    def test_archive_unarchive_compact_ok_and_audit(self):
+        task = self._create_task(executor="local")
+        self._set_status(task["id"], "failed")
+        for action in ("archive", "unarchive", "compact"):
+            status, body = self._req(
+                "POST", "/tasks/%s/thread/%s" % (task["id"], action),
+                {}, self.owner_key)
+            self.assertEqual(status, 200, body)
+        actions = self._audit_actions(self.project["id"])
+        self.assertIn("thread_archived", actions)
+        self.assertIn("thread_unarchived", actions)
+        self.assertIn("thread_compacted", actions)
+
+    def test_cli_failure_passthrough_502(self):
+        self._write_scenario(
+            {"fail_cmd": "thread-list", "fail_error": "list boom"})
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 502, body)
+        self.assertEqual(body["error"], "list boom")
+        self._write_scenario(
+            {"fail_cmd": "thread-archive", "fail_error": "arch boom"})
+        self._set_status(task["id"], "completed")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/archive" % task["id"], {},
+            self.owner_key)
+        self.assertEqual(status, 502, body)
+        self.assertEqual(body["error"], "arch boom")
+
+    def test_cli_bad_json_502(self):
+        self._write_scenario(
+            {"raw_stdout_for": "thread-list", "raw_stdout": "not json{"})
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 502, body)
+
+    def test_host_py_missing_502(self):
+        os.environ["CODEX_HOST_PY"] = os.path.join(
+            self.tmp.name, "does-not-exist.py")
+        task = self._create_task(executor="local")
+        status, body = self._req(
+            "GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 502, body)
+        self.assertIn("host.py", body["error"])
+
+
+if __name__ == "__main__":
+    unittest.main()