Skip to content

fix(chat): surface trigger_task dispatch failure + test env isolation #17

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
2233admin merged 1 commit into from
Jul 18, 2026
Merged

fix(chat): surface trigger_task dispatch failure + test env isolation #17

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions backend/api/v1/chat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -434,9 +434,25 @@ async def confirm(body: ConfirmRequest, db: AsyncSession = Depends(get_db)) -> A
await db.commit()
from backend.executor import get_executor

result = await get_executor().dispatch_collection(task.id, {})
try:
dispatch = await get_executor().dispatch_collection(task.id, {})
Comment on lines +437 to +438

@coderabbitai coderabbitai Bot Jul 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win

State loss and GC hazards in LocalExecutor.

While reviewing the get_executor().dispatch_collection() invocation, I noticed two critical concurrency and state issues in the LocalExecutor implementation provided in the context snippets:

  1. State loss in get_executor: get_executor() returns a new instance of LocalExecutor on every call. Because LocalExecutor relies on an instance-level _acquisition_tasks dictionary to track tasks, returning a new instance each time means previously dispatched tasks cannot be found or canceled later via cancel_acquisition.
  2. Task garbage collection hazard: LocalExecutor.dispatch_collection fires and forgets the collection task (t = asyncio.create_task(...)) without keeping a strong reference. The Python event loop only keeps weak references to tasks; this means active collection pipelines can be silently garbage-collected mid-execution when they yield to I/O.

Please consider applying @functools.lru_cache to get_executor in backend/executor/__init__.py to preserve the executor state, and updating LocalExecutor.dispatch_collection to track its spawned tasks in an instance dictionary (similar to _acquisition_tasks) to prevent them from vanishing.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/api/v1/chat.py` around lines 437 - 438, The executor factory
currently loses LocalExecutor state and dispatch_collection does not retain
spawned tasks. Cache get_executor so repeated calls return the same
LocalExecutor instance, and update LocalExecutor.dispatch_collection to store
each created asyncio task in an instance-level tracking dictionary, removing it
when it completes while preserving existing dispatch and cancellation behavior.

except Exception as exc:
# Task row is already committed; surface the dispatch failure instead
# of reporting applied=True with a silently dead task.
logger.exception("chat confirm | trigger_task dispatch failed source=%s task=%s", source.id, task.id)
raise HTTPException(
status_code=502, detail=f"任务已创建但派发失败 (task_id={task.id}), 请到工作项里重试"
) from exc
Comment on lines +439 to +445

@gemini-code-assist gemini-code-assist Bot Jul 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

当任务派发失败时，虽然 API 抛出了 502 异常，但由于 task 记录在前面已经通过 db.commit() 提交到了数据库，此时该任务在数据库中的状态仍然会保持为 pending。这会导致该任务在数据库和前端 UI 中永久处于“等待中”状态，给用户带来困惑。

建议在捕获到异常时，将 task.status 更新为 "failed" 并再次提交，以便准确反映任务的实际状态。

        except Exception as exc:
            # Task row is already committed; surface the dispatch failure instead
            # of reporting applied=True with a silently dead task.
            logger.exception("chat confirm | trigger_task dispatch failed source=%s task=%s", source.id, task.id)
            task.status = "failed"
            await db.commit()
            raise HTTPException(
                status_code=502, detail=f"任务已创建但派发失败 (task_id={task.id}), 请到工作项里重试"
            ) from exc

logger.info("chat confirm | trigger_task source=%s task=%s", source.id, task.id)
return ApiResponse.ok({"applied": True, "tool": proposal.tool, "task_id": task.id, "summary": proposal.summary})
return ApiResponse.ok(
{
"applied": True,
"tool": proposal.tool,
"task_id": task.id,
"summary": proposal.summary,
"dispatch": dispatch,
}
)

if proposal.tool == "update_schedule":
schedule = await schedule_service.get_schedule(db, args.get("schedule_id", ""))
Expand Down
53 changes: 53 additions & 0 deletions docs/BUG-TRIAGE-20260718.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
# Backend Bug Triage — 2026-07-18

部署验证 + 测试全量跑完后的后端问题账本。环境: main @ 50f48c8, Docker api+agent-1 (源码构建 0.3.6), host uv 环境跑测试。

## P0 — 结构性

### 1. identity 后端未进 main, production 前端登录死路
- 前端 `app/login/page.tsx` 三通道: OIDC (未配置) / bootstrap (`signInWithBootstrap` → `GET /api/v1/auth/me`) / 本地开发模式 (仅 `NODE_ENV !== 'production'`)
- main 后端 `backend/security/` 只有 `fleet_auth.py` + `url_guard.py`; `/auth/me`、`BOOTSTRAP_ADMIN_TOKEN` 的 identity 模块只存在于 `origin/codex/notification-ack`、`origin/codex/workflow-studio-motion-wip`
- 后果: `next build` 产物在 main 上**无法登录任何账号**。当前部署被迫用 `next dev` (工程模式) 绕过
- 修法: 把 codex 分支 identity 模块 (backend/security/identity.py + /auth/me 路由) 合回 main, 或前端登录页在 identity 后端缺席时降级

## P1 — 测试红 (真回归)

### 2. workflow import / demand-draft / turbopush 与 plan-IR 校验器脱节 (integration 9 挂)
- `tests/integration/test_workflow_patch_api.py` ×8 + `test_workflow_turbopush_publish_api.py` ×1
- 复现: `POST /api/v1/workflows/import/external-runtime` (langgraph 图) 返回 `valid: false`
- 根因错误码:
- `plan_ir_orphan_merge` — importer 把 langgraph `merge` 映射为 `intelligence.flow.merge` (要求 ≥2 入边), 导入图只有 1 入边
- `plan_ir_port_type_mismatch` — `external.tool.capability` 出口 `type='unknown'` 接不上 merge 入口 `recordCandidate[]`
- plan_ir 校验在 c42ece1 (four-level node hierarchy) 接入 compiler; importer (`backend/workflow/external_importer.py`) 没跟着更新映射
- 修向 (二选一): importer 合成合法图 (merge 补占位入边 / 外部工具出口给宽松类型), 或 plan-IR 对 external.* 目录节点放宽端口类型。牵涉工作流语义, 建议和工作流重构讨论一起定

## P2

### 3. chat confirm `trigger_task` 吞掉派发失败
- `backend/api/v1/chat.py:437`: `result = await get_executor().dispatch_collection(task.id, {})` 结果未检查, 派发炸了 API 仍回 `applied: true`
- 修法: 检查 result / try-except 回 `applied: false` + 原因
Comment on lines +26 to +28

@coderabbitai coderabbitai Bot Jul 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Mark the dispatch issue as resolved.

This entry still documents the pre-fix behavior (result unchecked and applied: true). The current implementation catches dispatch_collection failures and returns HTTP 502 with the committed task_id; update the triage record to reflect the shipped contract.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@docs/BUG-TRIAGE-20260718.md` around lines 26 - 28, Update the triage entry
for chat confirm trigger_task to mark the dispatch issue resolved and document
the shipped behavior: dispatch_collection failures are caught, the API returns
HTTP 502 with the committed task_id, and the prior unchecked-result/applied:true
behavior no longer applies.


## P3

### 4. 单测无 .env 隔离
- 根 `.env` 配了 `API_AUTH_TOKEN` (部署必需) 时, `tests/unit` 挂 19 个 (workers/nodes_install/geo 全是 401 或 token 注入断言)
- pydantic Settings 直读仓库根 `.env`; conftest 未清空鉴权相关 env
- 修法: conftest autouse fixture 强制 `API_AUTH_TOKEN=''`
Comment on lines +32 to +35

@coderabbitai coderabbitai Bot Jul 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Document both authentication variables in test isolation.

The remediation currently names only API_AUTH_TOKEN, but this PR also clears AGENT_API_TOKEN before backend import. Update the finding and fix description to cover both variables.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@docs/BUG-TRIAGE-20260718.md` around lines 32 - 35, Update the “单测无 .env 隔离”
finding and remediation text to document both authentication variables,
API_AUTH_TOKEN and AGENT_API_TOKEN. State that the conftest autouse fixture
clears both variables before backend imports, while preserving the existing
test-isolation context.


### 5. 杂项
- B904 raise-without-from 集中在 `backend/api/v1/browsers.py` (~8 处), 异常链丢失
- ruff 1701 条 (E501×679 / W293×388 为主, F841×17, F401×15); B008 是 FastAPI 惯用法, 建议 ruff config 加 per-file-ignores
- pytest-asyncio `event_loop_policy` fixture deprecation 警告 43 条

## 测试基线 (token env 清空后)

| 套件 | 结果 |
|---|---|
| tests/unit + tests/skills | 1355 passed, 5 skipped |
| tests/integration | 374 passed, 9 failed (上述 #2), 5 skipped |

复跑命令 (host):
```powershell
cd D:\projects\opencli-admin
$env:API_AUTH_TOKEN=''; $env:AGENT_API_TOKEN=''; uv run --extra dev pytest tests/unit tests/integration -q --no-cov
```
7 changes: 7 additions & 0 deletions tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,13 @@
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

# Deployment config in the repo-root .env must not leak into tests: with
# API_AUTH_TOKEN set, every unauthenticated test request would 401 (fleet
# auth, ADR-0005). Clear before backend.main import snapshots settings —
# an empty env var wins over the .env file value in pydantic-settings.
os.environ["API_AUTH_TOKEN"] = ""
os.environ["AGENT_API_TOKEN"] = ""

from backend.auth import crypto
from backend.database import Base, get_db
from backend.main import app
Expand Down
39 changes: 39 additions & 0 deletions tests/integration/test_chat_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,3 +130,42 @@ async def test_confirm_update_provider_not_found(client):
},
)
assert response.status_code == 404


# ── confirm: trigger_task dispatch failure ───────────────────────────────────
@pytest.mark.asyncio
async def test_confirm_trigger_task_reports_dispatch_failure(client, db_session, monkeypatch):
"""Dispatch blowing up after the task row is committed must surface as 502,
not applied=True with a silently dead task."""
from backend.models.source import DataSource

source = DataSource(
name="Chat Trigger Source",
channel_type="rss",
channel_config={"feed_url": "https://example.com/feed.xml"},
enabled=True,
)
db_session.add(source)
await db_session.commit()
await db_session.refresh(source)

class _BoomExecutor:
async def dispatch_collection(self, task_id: str, parameters: dict) -> dict:
raise RuntimeError("broker down")

monkeypatch.setattr("backend.executor.get_executor", lambda: _BoomExecutor())

response = await client.post(
"/api/v1/chat/confirm",
json={
"proposal": {
"tool": "trigger_task",
"args": {"source_id": source.id},
"summary": "触发采集",
"diff": "",
}
},
)

assert response.status_code == 502
assert "派发失败" in response.json()["detail"]
Comment on lines +170 to +171

@gemini-code-assist gemini-code-assist Bot Jul 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

为了确保在派发失败时，数据库中的任务状态确实被正确更新为了 "failed"，建议在测试中增加对任务状态的断言。

Suggested change
assert response.status_code == 502
assert "派发失败" in response.json()["detail"]
assert response.status_code == 502
assert "派发失败" in response.json()["detail"]
from backend.models.task import CollectionTask
from sqlalchemy import select
stmt = select(CollectionTask).where(CollectionTask.source_id == source.id)
db_task = (await db_session.execute(stmt)).scalar_one()
assert db_task.status == "failed"

Loading