From bae5ec3c58bdbf33f36e9f24fbf93a0ddb781b13 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Fri, 23 Oct 2026 15:57:33 +0800 Subject: [PATCH] F1: add correlated OpenCode HTTP and SSE task adapter --- backend/services/background_tasks/opencode.py | 146 ++++++++++++++++++ backend/tests/test_opencode_tasks.py | 54 +++++++ implement-stage/BUILD_NOTE.md | 3 +- 3 files changed, 202 insertions(+), 1 deletion(-) create mode 100644 backend/services/background_tasks/opencode.py create mode 100644 backend/tests/test_opencode_tasks.py diff --git a/backend/services/background_tasks/opencode.py b/backend/services/background_tasks/opencode.py new file mode 100644 index 0000000..30039ac --- /dev/null +++ b/backend/services/background_tasks/opencode.py @@ -0,0 +1,146 @@ +"""OpenCode HTTP/SSE adapter. No Pipecat, database or browser dependencies.""" + +import json +from collections.abc import AsyncIterator, Awaitable, Callable +from uuid import uuid4 + +import httpx + +Emit = Callable[[dict], Awaitable[None]] + + +class RemoteTaskError(RuntimeError): + """A remote task failed, without including connection credentials.""" + + +class OpenCodeExecutor: + def __init__(self, config: dict, secrets: dict, *, transport=None): + self.config = config + self._secrets = secrets + self._transport = transport + + def _client(self): + password = str(self._secrets.get("password") or "") + return httpx.AsyncClient( + base_url=self.config["url"].rstrip("/") + "/", + params={"directory": self.config["directory"]}, + auth=httpx.BasicAuth(self.config.get("username", "opencode"), password) if password else None, + timeout=httpx.Timeout(20, read=45), + follow_redirects=False, + transport=self._transport, + ) + + async def prepare(self) -> dict: + async with self._client() as client: + response = await client.post("session", json={"title": "语音助手后台任务"}) + response.raise_for_status() + session_id = response.json()["id"] + # Stored before prompt submission, so reconnects never create a second run. + return {"session_id": session_id, "message_id": "msg_" + uuid4().hex} + + @staticmethod + async def _snapshot(client, handle) -> dict: + response = await client.get(f"session/{handle['session_id']}/message") + response.raise_for_status() + messages = response.json() + status = await client.get("session/status") + status.raise_for_status() + remote_status = status.json().get(handle["session_id"], {"type": "idle"}) + matched = [item for item in messages if + item.get("info", {}).get("role") == "assistant" and + item.get("info", {}).get("parentID") == handle["message_id"]] + errors = [item["info"]["error"] for item in matched if item["info"].get("error")] + if errors: + return {"status": "failed", "summary": "远端执行失败", "result": json.dumps(errors[-1], ensure_ascii=False)[:8000]} + completed = [item for item in matched if item["info"].get("time", {}).get("completed")] + if remote_status.get("type") == "idle" and completed: + text = "\n".join(part.get("text", "") for item in completed for part in item.get("parts", []) if part.get("type") == "text") + return {"status": "succeeded", "summary": "任务已完成", "result": text[-16000:]} + return {"status": "running", "summary": "远端任务执行中", "result": ""} + + async def snapshot(self, handle: dict) -> dict: + async with self._client() as client: + return await self._snapshot(client, handle) + + async def run(self, handle: dict, instruction: str, emit: Emit, *, resume=False) -> dict: + """Subscribe before submitting; only correlated completed messages finish a run. + + Reconnect/restart never resubmits a prompt. The host calls this with + resume=True after a lost stream and reconciles from the messages API. + """ + async with self._client() as client: + if resume: + state = await self._snapshot(client, handle) + if state["status"] in {"succeeded", "failed"}: + return state + async with client.stream("GET", "event") as stream: + stream.raise_for_status() + if not resume: + response = await client.post( + f"session/{handle['session_id']}/prompt_async", + json={"messageID": handle["message_id"], "parts": [{"type": "text", "text": instruction}]}, + ) + response.raise_for_status() + await emit({"status": "submitted", "summary": "远端已接收，等待执行"}) + async for event in self._events(stream): + props = event.get("properties") or {} + info = props.get("info") or {} + part = props.get("part") or {} + session_id = props.get("sessionID") or info.get("sessionID") or part.get("sessionID") + if session_id != handle["session_id"]: + continue + kind = event.get("type") + if kind == "session.error": + return {"status": "failed", "summary": "远端执行失败", "result": "请检查 OpenCode 会话中的错误信息。"} + if kind in {"permission.asked", "question.asked"}: + request = {key: props[key] for key in ("id", "permission", "patterns", "questions") if key in props} + request["kind"] = "permission" if kind.startswith("permission") else "question" + await emit({"status": "waiting_input", "summary": "远端等待确认或补充信息", "input_request": request}) + elif kind == "message.part.updated" and part.get("type") == "tool": + state = part.get("state") or {} + await emit({"status": "running", "summary": f"{part.get('tool', '工具')} · {state.get('status', '执行中')}"[:300], "input_request": {}}) + elif kind == "session.status" and (props.get("status") or {}).get("type") == "busy": + await emit({"status": "running", "summary": "远端任务执行中"}) + if kind in {"session.idle", "session.status", "message.updated"}: + state = await self._snapshot(client, handle) + if state["status"] in {"succeeded", "failed"}: + return state + raise RemoteTaskError("远端事件流已关闭，需要重新连接") + + async def cancel(self, handle: dict) -> bool: + async with self._client() as client: + response = await client.post(f"session/{handle['session_id']}/abort") + response.raise_for_status() + status = await client.get("session/status") + status.raise_for_status() + value = status.json().get(handle["session_id"], {"type": "idle"}) + return value.get("type") == "idle" + + async def send_input(self, request: dict, value: dict): + async with self._client() as client: + if request.get("kind") == "permission": + reply = value.get("reply") + if reply not in {"once", "reject"}: + raise ValueError("请选择允许本次或拒绝") + response = await client.post(f"permission/{request['id']}/reply", json={"reply": reply}) + elif request.get("kind") == "question": + answers = value.get("answers") + if not isinstance(answers, list) or len(answers) != len(request.get("questions") or []): + raise ValueError("请回答远端提出的每一个问题") + response = await client.post(f"question/{request['id']}/reply", json={"answers": answers}) + else: + raise ValueError("当前任务没有待回答的问题") + response.raise_for_status() + + @staticmethod + async def _events(stream) -> AsyncIterator[dict]: + data = [] + async for line in stream.aiter_lines(): + if line.startswith("data:"): + data.append(line[5:].lstrip()) + elif not line and data: + try: + yield json.loads("\n".join(data)) + except json.JSONDecodeError: + pass + data = [] diff --git a/backend/tests/test_opencode_tasks.py b/backend/tests/test_opencode_tasks.py new file mode 100644 index 0000000..0e14cbb --- /dev/null +++ b/backend/tests/test_opencode_tasks.py @@ -0,0 +1,54 @@ +import json +import unittest +import httpx +from services.background_tasks.opencode import OpenCodeExecutor + + +class OpenCodeTaskTest(unittest.IsolatedAsyncioTestCase): + async def test_correlated_completion_ignores_initial_idle_and_other_sessions(self): + handle = {"session_id": "ses_test", "message_id": "msg_test"} + submitted = False + snapshots = 0 + updates = [] + async def respond(request): + nonlocal submitted, snapshots + self.assertEqual(request.url.params["directory"], "/workspace/demo") + if request.url.path == "/event": + events = [ + {"type": "session.idle", "properties": {"sessionID": "ses_other"}}, + {"type": "session.idle", "properties": {"sessionID": "ses_test"}}, + {"type": "message.updated", "properties": {"info": {"sessionID": "ses_test"}}}, + ] + return httpx.Response(200, text="".join("data: " + json.dumps(e) + "\n\n" for e in events)) + if request.url.path.endswith("prompt_async"): + self.assertFalse(submitted) + self.assertEqual(json.loads(request.content)["messageID"], "msg_test") + submitted = True + return httpx.Response(204) + if request.url.path.endswith("/message"): + snapshots += 1 + parent = "old_message" if snapshots == 1 else "msg_test" + return httpx.Response(200, json=[{"info": {"role": "assistant", "parentID": parent, "time": {"completed": 1}}, "parts": [{"type": "text", "text": "测试通过"}]}]) + if request.url.path == "/session/status": + return httpx.Response(200, json={}) + raise AssertionError(request.url) + executor = OpenCodeExecutor({"url": "https://remote.test", "directory": "/workspace/demo"}, {}, transport=httpx.MockTransport(respond)) + async def emit(event): + updates.append(event) + result = await executor.run(handle, "修复测试", emit) + self.assertEqual(result["status"], "succeeded") + self.assertEqual(result["result"], "测试通过") + self.assertEqual(snapshots, 2) + self.assertEqual(updates[0]["status"], "submitted") + + async def test_resume_reads_results_without_resubmitting(self): + async def respond(request): + self.assertEqual(request.method, "GET") + if request.url.path.endswith("/message"): + return httpx.Response(200, json=[{"info": {"role": "assistant", "parentID": "msg_test", "time": {"completed": 1}}, "parts": []}]) + return httpx.Response(200, json={}) + executor = OpenCodeExecutor({"url": "https://remote.test", "directory": "/demo"}, {}, transport=httpx.MockTransport(respond)) + async def emit(event): + self.fail("completed task should not restart") + state = await executor.run({"session_id": "ses_test", "message_id": "msg_test"}, "ignored", emit, resume=True) + self.assertEqual(state["status"], "succeeded") diff --git a/implement-stage/BUILD_NOTE.md b/implement-stage/BUILD_NOTE.md index 79d2fba..e41f37d 100644 --- a/implement-stage/BUILD_NOTE.md +++ b/implement-stage/BUILD_NOTE.md @@ -3,13 +3,14 @@ | Rung | Feature | Acceptance check | Tier | Status | |------|---------|------------------|------|--------| | F0 | 原生Job主线、断线独立与取消 | backend/.venv/bin/python -m unittest discover -s backend/tests -p test_background_task_native.py | MUST | passed | -| F1 | OpenCode协议适配与状态关联 | cd backend && .venv/bin/python -m unittest tests.test_background_task_native tests.test_opencode_tasks | MUST | pending | +| F1 | OpenCode协议适配与状态关联 | cd backend && .venv/bin/python -m unittest tests.test_background_task_native tests.test_opencode_tasks | MUST | passed | | F2 | 持久记录、管理API、应用宿主 | cd backend && .venv/bin/python -m unittest tests.test_background_tasks | MUST | pending | | F3 | 提示词绑定、任务通知、网页 | cd frontend && npx tsc --noEmit | MUST | pending | | F4 | 回归、文档与假设审阅 | make test | MUST | pending | ## Run record - F0: PYTHONPATH=backend backend/.venv/bin/python -m unittest discover -s backend/tests -p test_background_task_native.py → exit 0; native-jobs-smoke.json。 +- F1: 原生 Job + httpx 模拟远端协议共 3 项测试通过；未调用真实模型。 ## Deferred - 真实远程 OpenCode + 模型联调：未提供服务地址/凭证。 - 多进程分布式总线及通用插件发现：超出单进程 MVP。