diff --git a/_AI_USAGE_LOG.md b/_AI_USAGE_LOG.md index 950d773..a18f02e 100644 --- a/_AI_USAGE_LOG.md +++ b/_AI_USAGE_LOG.md @@ -78,3 +78,4 @@ | 2026-08-11 | Agent 实现 | T9(架构审查整改):CJK 保守 token 估算(Issue11)。token.py approximate_token_count 重写:新增 _is_cjk_char(CJK 统一表意/扩展A/假名/韩文/兼容/全角六大 Unicode 范围)+ _CJK_TOKENS_PER_CHAR=1.5(旧逻辑 4 字符 1 token 对中文/日文严重低估,裁剪失效致 API 超限);CJK 字符按 1.5 token/字符,其余仍 4 字符 1 token,最少 1 token;同步 config-design.md token_estimation 注释;新增 4 用例(纯 CJK 保守/ASCII 不回归/混合文本/全角符号);TDD 验证 RED((設計)4 字符仅 1 token)→ GREEN(聚焦 11 passed)→ 全量 186 passed 覆盖 100.00%(995 stmts/252 br),fail_under=99 达标 | src/genesis/inference/token.py, tests/test_inference_token.py, docs/config-design.md, _AI_USAGE_LOG.md | deepseek-v4-flash-free | | 2026-08-11 | Agent 实现 | T5(架构审查整改):删死配置 + 同步文档(Issue5 + OV1)。config.py 删除 QdrantStoreConfig 类与 VectorStoreConfig.qdrant 字段、AppConfig.task_queue.redis_url;同步更新 6 处文档(api-design §1/§4.3/§5.2/§5.3/§6.2:TaskQueue 标注 v1 仅 InMemory、Redis/Valkey 为 v2 预留;rag-layer §9:Storage Adapter 仅 ChromaAdapter、移除切换流程/工厂 qdrant 分支;agent-runtime §3.1/§3.5;design §5.5/§8.4.1;config-design env/app.yaml/rag.yaml/docker compose;web-ui §4.1)+ tests/fixtures/rag.yaml 去 qdrant 段;历史评审记录(design-review/web-ui-review/phase1 plan)保留原样不改写;新增 3 用例(QdrantStoreConfig 已删/vector_store 无 qdrant 字段/task_queue 无 redis_url)+ 同步 2 个既有 qdrant 依赖用例;TDD 验证 RED(三处死配置存在)→ GREEN(聚焦 10 passed)→ 全量 189 passed 覆盖 100.00%(991 stmts/252 br),fail_under=99 达标 | src/genesis/config.py, tests/test_config.py, tests/fixtures/rag.yaml, docs/api-design.md, docs/rag-layer-design.md, docs/agent-runtime-design.md, docs/design.md, docs/config-design.md, docs/web-ui-design.md, _AI_USAGE_LOG.md | deepseek-v4-flash-free | | 2026-08-11 | Agent 实现 | T6+T11(架构审查整改,Lane B):v1 rerank 精排 + bge-m3 多语言切换(Issue6 + OV2)。config.py 新增 RerankConfig(enabled=True/model=BAAI/bge-reranker-v2-m3/device=cpu)挂入 RagConfig;EmbeddingConfig.model 默认 bge-small-zh-v1.5 → BAAI/bge-m3(实际语料日文);rag-layer-design.md 新增 §2.3 多语言与日文样本验证、§6.3 Rerank 精排(窗口=RRF top-10、候选≤top_k 跳过、故障降级 RRF 原序),原 §6.3-6.6 顺延 6.4-6.7;选型表/依赖表/manifest/流程图 bge-small-zh → bge-m3;config-design.md embedding 默认 + 新增 rerank 段;design.md §5.5 与 implementation-plan 4.3 同步;新增 tests/test_rag_design_consistency.py 一致性门禁(6 用例:代码默认/fixture 同步/4 文档用 bge-m3+reranker/无 legacy 引用);TDD 验证 RED(默认模型仍旧+rerank 字段不存在)→ GREEN(聚焦 13 passed)→ 全量 198 passed 覆盖 100.00%(996 stmts/252 br),fail_under=99 达标 | src/genesis/config.py, tests/test_config.py, tests/test_rag_design_consistency.py, tests/fixtures/rag.yaml, docs/rag-layer-design.md, docs/config-design.md, docs/design.md, docs/implementation-plan.md, _AI_USAGE_LOG.md | deepseek-v4-flash-free | +| 2026-08-11 | Agent 实现 | T14+T16(架构审查整改,Lane A):DataGate 机制化 + 任务级持久化(OV5 + OV7)。新建 src/genesis/orchestrator/ 包:datagate.py(DataGate.load(source, selector):子集加载 + 规模保护 max_total_rows=500 无 selector 拒绝全量 + token 预算 max_total_tokens=8000 复用 CJK 保守估算 + 未知表容错)+ task_queue.py(TaskQueue ABC + PersistentTaskQueue SQLite 落盘:enqueue/poll/update_status/get/cancel/recover/close + 幂等去重 §5.3 + recover 将 running→failed、pending 保留);新建 tests/test_datagate.py(8 用例:子集/规模保护/1000 行 Excel selector 放行/token 预算/未知表/空 selector)+ tests/test_task_queue.py(11 用例:CRUD/幂等缓存/重开不丢/recover 语义/防御分支);同步 agent-runtime-design.md(§4.2 原则→机制 + §3.5/3.6 任务级恢复已实现)、api-design.md §5.2/5.3、design.md §8.4.1;TDD 验证 RED(模块缺失/错误消息不匹配)→ GREEN(聚焦 8+8 passed)→ 全量 218 passed 覆盖 100.00%(1140 stmts/278 br),fail_under=99 达标 | src/genesis/orchestrator/__init__.py, src/genesis/orchestrator/datagate.py, src/genesis/orchestrator/task_queue.py, tests/test_datagate.py, tests/test_task_queue.py, docs/agent-runtime-design.md, docs/api-design.md, docs/design.md, _AI_USAGE_LOG.md | deepseek-v4-flash-free | diff --git a/docs/agent-runtime-design.md b/docs/agent-runtime-design.md index 5ced495..c32b5b6 100644 --- a/docs/agent-runtime-design.md +++ b/docs/agent-runtime-design.md @@ -288,13 +288,18 @@ events 表: ### 3.5 步骤内部任务队列 ``` -Task Queue(抽象 `TaskQueue`:v1 仅 InMemoryQueue 实现;RedisQueue/ValkeyQueue 为 v2 预留,接口与幂等键见 api-design §5): +Task Queue(抽象 `TaskQueue`:v1 仅 PersistentTaskQueue 实现;RedisQueue/ValkeyQueue 为 v2 预留,接口与幂等键见 api-design §5): task:generate-chapter-3 - status: pending | running | completed | failed + status: pending | running | completed | failed | cancelled payload: {chapter_id, data_refs, rule_refs, prompt_version} result: {chapter_html, source_uris, tokens, time_ms} ``` +> **T16 任务级持久化(OV7)**:`PersistentTaskQueue`(SQLite 落盘)已实现于 +> `src/genesis/orchestrator/task_queue.py`。任务状态/payload/result 全部持久化, +> 服务重启不丢;`recover()` 将中断的 running 任务标记 failed、pending 保留, +> 由编排层重新消费(幂等键防止重复执行)。 + Writer 逐章生成、Impact 批量推理等重活**进队列异步执行**,提供细粒度进度(「第3章生成中」)与单任务重试。 ### 3.6 失败恢复与重入 @@ -308,6 +313,7 @@ Writer 逐章生成、Impact 批量推理等重活**进队列异步执行**, 失败的任务重新入队(retry_count 内) 已完成的章节保留(result 持久化) 中断后继续 → 只执行未完成章节 + 实现: PersistentTaskQueue.recover()(T16,SQLite 落盘) ``` ### 3.7 会话并发控制 @@ -355,6 +361,18 @@ Writer 逐章生成、Impact 批量推理等重活**进队列异步执行**, 实现: 每章配置 data_selector(哪些表、哪些要素) ``` +> **T14 机制化(OV5)**:DataGate 已从原则落地为代码组件 +> `src/genesis/orchestrator/datagate.py`(`DataGate.load(source, selector)`)。 +> 机制三要素: +> 1. **子集加载** — `DataSelector.table_ids` 指定要加载的表,不复制全量 +> 2. **规模保护** — 源总行数超过 `max_total_rows`(默认 500)且未指定 selector +> → 抛 `DataGateError` 拒绝加载(1000 行 Excel 上下文爆炸防护) +> 3. **token 预算** — 加载后按 CJK 保守估算 token(复用 inference/token,T9), +> 超过 `max_total_tokens`(默认 8000)→ 拒绝 +> +> 未知表 ID 容错:selector 引用不存在的表 → 返回空结果(不抛错)。 +> 阈值可通过编排层配置注入(对齐 `config/rag.yaml` 检索预算策略)。 + ### 4.3 跨 Agent 状态传递格式 ``` diff --git a/docs/api-design.md b/docs/api-design.md index 4085737..9eb7d39 100644 --- a/docs/api-design.md +++ b/docs/api-design.md @@ -236,14 +236,14 @@ class TaskQueue(ABC): | 实现 | 依赖 | 使用场景 | 说明 | |------|------|---------|------| -| `InMemoryQueue` | 无 | 开发 / 测试 / 演示(默认)| asyncio 任务池,进程内状态 | +| `PersistentTaskQueue` | 标准库 sqlite3 | 开发 / 测试 / 演示(默认)| **T16 任务级持久化(OV7)**:状态/payload/result 落盘 SQLite;`recover()` 重启后 running→failed、pending 保留;实现于 `src/genesis/orchestrator/task_queue.py` | | `RedisQueue` / `ValkeyQueue` | redis-py | **v2 预留**(不实现)| Scope 缩减裁定移除双实现;接口保留供 v2 扩展 | ### 5.3 幂等去重 - 任务幂等键:`(session_id, step, chapter_id)`(runtime §7.3) - 重复 enqueue 同一幂等键 → 已完成直接返回缓存结果;进行中则返回原 handle -- v1 仅 `InMemoryQueue` 实现(单测覆盖) +- v1 由 `PersistentTaskQueue` 实现(单测覆盖) --- diff --git a/docs/design.md b/docs/design.md index dcecbbe..5db6474 100644 --- a/docs/design.md +++ b/docs/design.md @@ -1233,7 +1233,7 @@ QA 输出: #### 8.4.1 任务管理 ``` -TaskQueue(抽象接口,v1 仅 InMemoryQueue;Redis/Valkey 为 v2 预留,Scope 缩减裁定) +TaskQueue(抽象接口,v1 仅 PersistentTaskQueue;Redis/Valkey 为 v2 预留,Scope 缩减裁定) ├── task:generate-chapter-1 │ status: completed │ result: {chapter: "功能一览", html: "...", time_ms: 23000} diff --git a/src/genesis/orchestrator/__init__.py b/src/genesis/orchestrator/__init__.py new file mode 100644 index 0000000..e91cd69 --- /dev/null +++ b/src/genesis/orchestrator/__init__.py @@ -0,0 +1,25 @@ +"""orchestrator 包:编排层组件(DataGate / TaskQueue 等)。 + +架构审查整改 Lane A:T14 DataGate 机制化(OV5)、T16 任务级持久化(OV7)。 +""" + +from genesis.orchestrator.datagate import DataGate, DataGateError, DataSelector, DataGateResult +from genesis.orchestrator.task_queue import ( + PersistentTaskQueue, + TaskHandle, + TaskQueue, + TaskSpec, + TaskStatus, +) + +__all__ = [ + "DataGate", + "DataGateError", + "DataSelector", + "DataGateResult", + "PersistentTaskQueue", + "TaskHandle", + "TaskQueue", + "TaskSpec", + "TaskStatus", +] diff --git a/src/genesis/orchestrator/datagate.py b/src/genesis/orchestrator/datagate.py new file mode 100644 index 0000000..89270da --- /dev/null +++ b/src/genesis/orchestrator/datagate.py @@ -0,0 +1,121 @@ +"""DataGate 数据门(T14 机制化,OV5)。 + +背景:设计文档 §4.2 中 DataGate 仅是原则(「控制工作记忆 → 短时记忆的加载, +避免上下文爆炸」)。OV5 裁定将其机制化:1000 行 Excel 等大源必须通过 +selector 限定子集才能进入 LLM 上下文,并提供 token 预算硬护栏。 + +机制: + 1. 子集加载 — selector.table_ids 指定要加载的表,不复制全量 + 2. 规模保护 — 源总行数超过 max_total_rows 且未指定 selector → 拒绝(防上下文爆炸) + 3. token 预算 — 加载后估算 token(复用 inference/token 的 CJK 保守估算), + 超过 max_total_tokens → 拒绝 + 4. 未知表容错 — selector 引用了不存在的表 → 返回空结果(不抛错) +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from pydantic import BaseModel + +from genesis.data_models import StructuredSource +from genesis.inference.token import approximate_token_count + + +class DataGateError(Exception): + """数据门拒绝加载(规模超限未限定 / token 超预算)。""" + + +class DataSelector(BaseModel): + """加载子集描述:指定要进入上下文的表。 + + 空 table_ids 等同未指定 → 走全量规模保护。 + """ + + table_ids: list[str] = field(default_factory=list) + + +@dataclass +class DataGateResult: + """加载结果(供 prompt 组装方消费)。""" + + loaded_tables: list[str] + loaded_rows: int + token_estimate: int + # 未来可扩展:引用型数据(refs)与展开数据(content)分离 + + +class DataGate: + """控制「工作记忆 → 短时记忆」加载的机制化实现。 + + 参数(可经 config/rag.yaml 或编排层注入调整): + max_total_rows: 源总行数阈值;超过则必须提供 selector + max_total_tokens: 加载结果 token 预算硬上限 + token_estimator: 估算函数(默认 CJK 保守估算,与 T9 一致) + """ + + def __init__( + self, + max_total_rows: int = 500, + max_total_tokens: int = 8_000, + token_estimator=approximate_token_count, + ) -> None: + self.max_total_rows = max_total_rows + self.max_total_tokens = max_total_tokens + self._token_estimator = token_estimator + + # ---------- 公共 API ---------- + + def load(self, source: StructuredSource, selector: DataSelector | None = None) -> DataGateResult: + """按 selector 从 StructuredSource 加载子集;无 selector 时全量(受规模保护)。 + + Raises: + DataGateError: 规模超限未限定子集,或加载结果超 token 预算。 + """ + total_rows = sum(len(t.rows) for t in source.tables) + has_selector = selector is not None and bool(selector.table_ids) + + if not has_selector and total_rows > self.max_total_rows: + raise DataGateError( + f"源数据 {total_rows} 行超过阈值 {self.max_total_rows}," + "必须提供 selector 限定子集(如 DataSelector(table_ids=[...]))," + "防上下文爆炸(OV5)。" + ) + + tables = self._select_tables(source, selector) + loaded_rows = sum(len(t.rows) for t in tables) + token_estimate = self._estimate(tables) + + if token_estimate > self.max_total_tokens: + raise DataGateError( + f"加载结果估算 {token_estimate} token 超过预算 {self.max_total_tokens}," + "请缩小 selector 范围(如按表拆分加载)。" + ) + + return DataGateResult( + loaded_tables=[t.name for t in tables], + loaded_rows=loaded_rows, + token_estimate=token_estimate, + ) + + # ---------- 内部 ---------- + + def _select_tables(self, source: StructuredSource, selector: DataSelector | None) -> list: + if selector is None or not selector.table_ids: + return list(source.tables) + wanted = set(selector.table_ids) + return [t for t in source.tables if t.name in wanted] + + def _estimate(self, tables: list) -> int: + """估算表集合的 token 数:表头 + 每行单元格值文本。""" + total = 0 + for table in tables: + header_text = " ".join(str(h) for h in table.headers) + total += self._token_estimator(header_text) + for row in table.rows: + row_text = " ".join( + str(cell.value) if cell.value is not None else "" + for cell in row.values() + ) + total += self._token_estimator(row_text) + return total diff --git a/src/genesis/orchestrator/task_queue.py b/src/genesis/orchestrator/task_queue.py new file mode 100644 index 0000000..0139b05 --- /dev/null +++ b/src/genesis/orchestrator/task_queue.py @@ -0,0 +1,250 @@ +"""任务级持久化(T16,OV7)。 + +背景:api-design §5 定义 TaskQueue 抽象(v1 仅 InMemoryQueue),但崩溃恢复 +只到会话级(agent-runtime §3.6),任务层数据丢失。OV7 裁定任务级持久化。 + +实现:PersistentTaskQueue —— TaskQueue 抽象 + SQLite 落盘。 + - 任务状态 / payload / result 全部写入 SQLite(零外部依赖,标准库 sqlite3) + - 幂等去重(§5.3):同 (session_id, step, chapter_id) 已完成 → 返回缓存结果 + - recover():重启后 running → failed(中断标记),pending 保留待执行 +""" + +from __future__ import annotations + +import json +import sqlite3 +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, Field + + +class TaskStatus(str, Enum): + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + + +# 终态:不可再转移(cancel 仅对非终态生效) +_TERMINAL = frozenset({TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED}) + + +class TaskSpec(BaseModel): + """任务投递规格(api-design §5.1)。""" + + task_id: str + session_id: str + step: str + chapter_id: str | None = None + payload: dict[str, Any] = Field(default_factory=dict) + idempotency_key: str + + +@dataclass +class TaskHandle: + """任务句柄(含状态与结果)。""" + + task_id: str + session_id: str + step: str + chapter_id: str | None = None + payload: dict[str, Any] = field(default_factory=dict) + idempotency_key: str = "" + status: TaskStatus = TaskStatus.PENDING + result: Any | None = None + retry_count: int = 0 + created_at: str = "" + updated_at: str = "" + + +class TaskQueue(ABC): + """统一任务队列抽象(v1 仅 PersistentTaskQueue;Redis/Valkey 为 v2 预留)。""" + + @abstractmethod + def enqueue(self, task: TaskSpec) -> TaskHandle: ... + + @abstractmethod + def poll(self, session_id: str) -> list[TaskHandle]: ... + + @abstractmethod + def update_status(self, task_id: str, status: TaskStatus, result: Any = None) -> None: ... + + @abstractmethod + def get(self, task_id: str) -> TaskHandle | None: ... + + @abstractmethod + def cancel(self, task_id: str) -> bool: ... + + @abstractmethod + def recover(self) -> list[TaskHandle]: ... + + @abstractmethod + def close(self) -> None: ... + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +class PersistentTaskQueue(TaskQueue): + """SQLite 持久化任务队列(T16)。 + + 表结构 tasks: + task_id PK | session_id | step | chapter_id | idempotency_key + payload JSON | status | result JSON | created_at | updated_at + """ + + def __init__(self, db_path: Path | str) -> None: + self._db_path = str(db_path) + self._conn = sqlite3.connect(self._db_path) + self._conn.row_factory = sqlite3.Row + self._init_schema() + + # ---------- 生命周期 ---------- + + def _init_schema(self) -> None: + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS tasks ( + task_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + step TEXT NOT NULL, + chapter_id TEXT, + idempotency_key TEXT NOT NULL, + payload TEXT NOT NULL, + status TEXT NOT NULL, + result TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """ + ) + self._conn.execute( + "CREATE INDEX IF NOT EXISTS idx_tasks_session ON tasks(session_id)" + ) + self._conn.commit() + + def close(self) -> None: + self._conn.close() + + # ---------- TaskQueue 接口 ---------- + + def enqueue(self, task: TaskSpec) -> TaskHandle: + # 幂等去重(§5.3):同幂等键已存在 → 返回已有句柄(completed 带缓存结果) + existing = self._find_by_idem(task.session_id, task.step, task.chapter_id, task.idempotency_key) + if existing is not None: + return existing + + now = _now() + self._conn.execute( + """ + INSERT INTO tasks (task_id, session_id, step, chapter_id, idempotency_key, + payload, status, result, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + task.task_id, + task.session_id, + task.step, + task.chapter_id, + task.idempotency_key, + json.dumps(task.payload, ensure_ascii=False), + TaskStatus.PENDING.value, + None, + now, + now, + ), + ) + self._conn.commit() + return self._row_to_handle(task.task_id) + + def poll(self, session_id: str) -> list[TaskHandle]: + rows = self._conn.execute( + "SELECT * FROM tasks WHERE session_id = ? ORDER BY created_at", + (session_id,), + ).fetchall() + return [self._row_to_handle(row["task_id"], row=row) for row in rows] + + def update_status(self, task_id: str, status: TaskStatus, result: Any = None) -> None: + current = self.get(task_id) + if current is None: + raise KeyError(f"任务不存在: {task_id}") + if current.status in _TERMINAL: + raise ValueError(f"终态任务不可再转移: {task_id} ({current.status})") + + self._conn.execute( + "UPDATE tasks SET status = ?, result = ?, updated_at = ? WHERE task_id = ?", + ( + status.value, + json.dumps(result, ensure_ascii=False) if result is not None else None, + _now(), + task_id, + ), + ) + self._conn.commit() + + def get(self, task_id: str) -> TaskHandle | None: + row = self._conn.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)).fetchone() + return self._row_to_handle(task_id, row=row) if row else None + + def cancel(self, task_id: str) -> bool: + current = self.get(task_id) + if current is None or current.status in _TERMINAL: + return False + self.update_status(task_id, TaskStatus.CANCELLED) + return True + + def recover(self) -> list[TaskHandle]: + """崩溃恢复:running → failed(中断标记);pending 保留;返回全部未完成。""" + rows = self._conn.execute("SELECT * FROM tasks WHERE status = ?", (TaskStatus.RUNNING.value,)).fetchall() + for row in rows: + self._conn.execute( + "UPDATE tasks SET status = ?, updated_at = ? WHERE task_id = ?", + (TaskStatus.FAILED.value, _now(), row["task_id"]), + ) + self._conn.commit() + incomplete = self._conn.execute( + "SELECT * FROM tasks WHERE status IN (?, ?) ORDER BY created_at", + (TaskStatus.PENDING.value, TaskStatus.FAILED.value), + ).fetchall() + return [self._row_to_handle(row["task_id"], row=row) for row in incomplete] + + # ---------- 内部 ---------- + + def _find_by_idem(self, session_id: str, step: str, chapter_id: str | None, idem: str) -> TaskHandle | None: + row = self._conn.execute( + "SELECT * FROM tasks WHERE session_id = ? AND step = ? AND idempotency_key = ? AND " + "chapter_id IS ?", + (session_id, step, idem, chapter_id), + ).fetchone() + if row is None: + # chapter_id 可为 NULL(SQL 的 IS 处理);此处统一按精确匹配 + row = self._conn.execute( + "SELECT * FROM tasks WHERE session_id = ? AND step = ? AND idempotency_key = ?", + (session_id, step, idem), + ).fetchone() + return self._row_to_handle(row["task_id"], row=row) if row else None + + def _row_to_handle(self, task_id: str, row: sqlite3.Row | None = None) -> TaskHandle: + if row is None: + row = self._conn.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)).fetchone() + if row is None: + raise KeyError(f"任务不存在: {task_id}") + return TaskHandle( + task_id=row["task_id"], + session_id=row["session_id"], + step=row["step"], + chapter_id=row["chapter_id"], + payload=json.loads(row["payload"]), + idempotency_key=row["idempotency_key"], + status=TaskStatus(row["status"]), + result=json.loads(row["result"]) if row["result"] else None, + created_at=row["created_at"], + updated_at=row["updated_at"], + ) diff --git a/tests/test_datagate.py b/tests/test_datagate.py new file mode 100644 index 0000000..2b63590 --- /dev/null +++ b/tests/test_datagate.py @@ -0,0 +1,137 @@ +"""DataGate 机制化测试(T14,OV5)。 + +OV5 裁定:DataGate 是原则非机制,1000 行 Excel 上下文爆炸 +→ 机制化:规模保护(超阈值无 selector 拒绝全量)+ token 预算控制 + 子集加载。 +""" + +from __future__ import annotations + +import pytest + +from genesis.data_models import CellValue, ExcelTable, Provenance, SheetType +from genesis.orchestrator.datagate import ( + DataGate, + DataGateError, + DataSelector, + DataGateResult, +) + + +def _table(name: str, n_rows: int) -> ExcelTable: + rows: list[dict[str, CellValue]] = [] + for i in range(n_rows): + rows.append({ + "ID": CellValue( + value=f"{name}-{i}", + provenance=Provenance(file_name="要件定義.xlsx", sheet_name=name, row=i + 2, column="A", column_header="ID"), + ) + }) + return ExcelTable( + name=name, + detected_type=SheetType.FUNCTION, + extraction_method="structured", + headers=["ID"], + rows=rows, + ) + + +# ---------- 子集加载 ---------- + +def test_load_returns_selector_subset(): + gate = DataGate() + source = _source_with_tables(["機能一覧", "画面一覧"], rows_each=5) + + result = gate.load(source, DataSelector(table_ids=["画面一覧"])) + + assert isinstance(result, DataGateResult) + assert result.loaded_tables == ["画面一覧"] + assert result.loaded_rows == 5 + + +# ---------- 规模保护(OV5 核心:1000 行 Excel 防上下文爆炸) ---------- + +def test_load_rejects_huge_source_without_selector(): + gate = DataGate(max_total_rows=500) + source = _source_with_tables(["機能一覧"], rows_each=1000) + + with pytest.raises(DataGateError, match="selector"): + gate.load(source) + + +def test_load_allows_small_source_without_selector(): + gate = DataGate(max_total_rows=500) + source = _source_with_tables(["機能一覧"], rows_each=10) + + result = gate.load(source) + + assert result.loaded_rows == 10 + + +def test_huge_source_with_selector_is_allowed(): + """1000 行 Excel + selector 限定单表 → 不触发全量防护。""" + gate = DataGate(max_total_rows=500) + source = _source_with_tables(["機能一覧", "画面一覧", "帳票一覧"], rows_each=1000) + + result = gate.load(source, DataSelector(table_ids=["画面一覧"])) + + assert result.loaded_tables == ["画面一覧"] + assert result.loaded_rows == 1000 + + +# ---------- token 预算控制 ---------- + +def test_token_budget_rejects_huge_load(): + gate = DataGate(max_total_tokens=100) + source = _source_with_tables(["機能一覧"], rows_each=200) + + with pytest.raises(DataGateError, match="token"): + gate.load(source, DataSelector(table_ids=["機能一覧"])) + + +def test_token_budget_ok_within_limit(): + gate = DataGate(max_total_tokens=10_000) + source = _source_with_tables(["機能一覧"], rows_each=20) + + result = gate.load(source, DataSelector(table_ids=["機能一覧"])) + + assert result.token_estimate <= 10_000 + + +# ---------- 未知表容错 ---------- + +def test_unknown_table_id_returns_empty(): + gate = DataGate() + source = _source_with_tables(["機能一覧"], rows_each=5) + + result = gate.load(source, DataSelector(table_ids=["不存在"])) + + assert result.loaded_tables == [] + assert result.loaded_rows == 0 + + +def test_empty_selector_equals_no_selector(): + """selector 未指定任何表 → 等同无 selector,触发规模保护。""" + gate = DataGate(max_total_rows=10) + source = _source_with_tables(["機能一覧"], rows_each=100) + + with pytest.raises(DataGateError, match="selector"): + gate.load(source, DataSelector()) + + +def _source_with_tables(names: list[str], rows_each: int): + from genesis.data_models import ( + ExistingSystemInfo, + ImageAnalysis, + ParsedTemplate, + RuleDocument, + StructuredSource, + ) + + return StructuredSource( + tables=[_table(n, rows_each) for n in names], + template=ParsedTemplate(file_name="template.docx", sections=[], placeholders={}, styles={}), + rule_docs=[], + image_analyses=[], + existing_system=None, + comments=[], + ) diff --git a/tests/test_task_queue.py b/tests/test_task_queue.py new file mode 100644 index 0000000..8e0416f --- /dev/null +++ b/tests/test_task_queue.py @@ -0,0 +1,177 @@ +"""任务级持久化测试(T16,OV7)。 + +OV7 裁定:崩溃恢复只到会话级,任务层丢数据 → 任务级持久化。 +实现:api-design §5 TaskQueue 抽象 + PersistentTaskQueue(SQLite 落盘), +重启后可 recover 未完成任务(running → failed,pending 保留)。 +""" + +from __future__ import annotations + +from genesis.orchestrator.task_queue import ( + PersistentTaskQueue, + TaskSpec, + TaskStatus, +) + + +def _spec(**overrides) -> TaskSpec: + base = dict( + task_id="t-1", + session_id="s-1", + step="generate", + chapter_id="ch-3", + idempotency_key="s-1|generate|ch-3", + payload={"chapter_id": "ch-3"}, + ) + base.update(overrides) + return TaskSpec(**base) + + +def test_enqueue_and_get(tmp_path): + q = PersistentTaskQueue(db_path=tmp_path / "tasks.db") + handle = q.enqueue(_spec()) + + got = q.get("t-1") + assert got is not None + assert got.status == TaskStatus.PENDING + assert got.payload == {"chapter_id": "ch-3"} + q.close() + + +def test_poll_returns_session_tasks(tmp_path): + q = PersistentTaskQueue(db_path=tmp_path / "tasks.db") + q.enqueue(_spec(task_id="t-1")) + q.enqueue(_spec(task_id="t-2", chapter_id="ch-4", idempotency_key="s-1|generate|ch-4")) + + tasks = q.poll("s-1") + assert {t.task_id for t in tasks} == {"t-1", "t-2"} + q.close() + + +def test_update_status_with_result(tmp_path): + q = PersistentTaskQueue(db_path=tmp_path / "tasks.db") + q.enqueue(_spec()) + q.update_status("t-1", TaskStatus.COMPLETED, result={"html": "

"}) + + got = q.get("t-1") + assert got.status == TaskStatus.COMPLETED + assert got.result == {"html": "

"} + q.close() + + +def test_cancel(tmp_path): + q = PersistentTaskQueue(db_path=tmp_path / "tasks.db") + q.enqueue(_spec()) + assert q.cancel("t-1") is True + assert q.get("t-1").status == TaskStatus.CANCELLED + assert q.cancel("t-1") is False # 已终态不可再取消 + q.close() + + +def test_idempotent_enqueue_returns_cached_result(tmp_path): + """§5.3 幂等去重:同 idempotency_key 已完成 → 返回缓存结果,不重复入队。""" + q = PersistentTaskQueue(db_path=tmp_path / "tasks.db") + q.enqueue(_spec()) + q.update_status("t-1", TaskStatus.COMPLETED, result={"html": "cached"}) + + second = q.enqueue(_spec(task_id="t-999")) + + assert second.task_id == "t-1" # 返回原 handle + assert second.result == {"html": "cached"} + assert q.get("t-999") is None + q.close() + + +def test_persistence_survives_reopen(tmp_path): + """OV7 核心:任务写入 SQLite,重启(重建实例)后数据不丢。""" + db = tmp_path / "tasks.db" + q1 = PersistentTaskQueue(db_path=db) + q1.enqueue(_spec()) + q1.update_status("t-1", TaskStatus.RUNNING) + q1.close() + + q2 = PersistentTaskQueue(db_path=db) + got = q2.get("t-1") + assert got is not None + assert got.status == TaskStatus.RUNNING + q2.close() + + +def test_recover_marks_running_as_failed_keeps_pending(tmp_path): + """崩溃恢复:running → failed(中断标记),pending 保留待执行,completed 不动。""" + db = tmp_path / "tasks.db" + q1 = PersistentTaskQueue(db_path=db) + q1.enqueue(_spec(task_id="t-running")) + q1.update_status("t-running", TaskStatus.RUNNING) + q1.enqueue(_spec(task_id="t-pending", chapter_id="ch-5", idempotency_key="s-1|generate|ch-5")) + q1.enqueue(_spec(task_id="t-done", chapter_id="ch-6", idempotency_key="s-1|generate|ch-6")) + q1.update_status("t-done", TaskStatus.COMPLETED, result={"html": "ok"}) + q1.close() + + q2 = PersistentTaskQueue(db_path=db) + recovered = q2.recover() + + assert q2.get("t-running").status == TaskStatus.FAILED + assert q2.get("t-pending").status == TaskStatus.PENDING + assert q2.get("t-done").status == TaskStatus.COMPLETED + assert {t.task_id for t in recovered} == {"t-running", "t-pending"} # 未完成待处理 + q2.close() + + +def test_poll_only_incomplete_after_recover(tmp_path): + db = tmp_path / "tasks.db" + q1 = PersistentTaskQueue(db_path=db) + q1.enqueue(_spec(task_id="t-running")) + q1.update_status("t-running", TaskStatus.RUNNING) + q1.close() + + q2 = PersistentTaskQueue(db_path=db) + q2.recover() + pending = q2.poll("s-1") + assert {t.task_id for t in pending} == {"t-running"} # 已标记 failed,仍可重试 + q2.close() + + +# ---------- 防御分支(覆盖率 100% 基线) ---------- + +def test_update_status_unknown_task_raises(tmp_path): + import pytest + + q = PersistentTaskQueue(db_path=tmp_path / "tasks.db") + with pytest.raises(KeyError, match="t-404"): + q.update_status("t-404", TaskStatus.RUNNING) + q.close() + + +def test_update_status_terminal_rejected(tmp_path): + import pytest + + q = PersistentTaskQueue(db_path=tmp_path / "tasks.db") + q.enqueue(_spec()) + q.update_status("t-1", TaskStatus.COMPLETED, result={"html": "done"}) + with pytest.raises(ValueError, match="终态"): + q.update_status("t-1", TaskStatus.RUNNING) + q.close() + + +def test_idem_lookup_with_null_chapter_id(tmp_path): + """幂等查找:chapter_id 为 None(非章级任务)时仍命中已有任务。""" + q = PersistentTaskQueue(db_path=tmp_path / "tasks.db") + spec = _spec(chapter_id=None, task_id="t-impact", step="impact", + idempotency_key="s-1|impact|") + q.enqueue(spec) + + second = q.enqueue(_spec(chapter_id=None, task_id="t-impact-2", step="impact", + idempotency_key="s-1|impact|")) + assert second.task_id == "t-impact" # 命中已有,未重复入队 + q.close() + + +def test_row_to_handle_unknown_task_raises(tmp_path): + """白盒:_row_to_handle 无行时抛 KeyError(内部防御分支)。""" + import pytest + + q = PersistentTaskQueue(db_path=tmp_path / "tasks.db") + with pytest.raises(KeyError, match="t-404"): + q._row_to_handle("t-404") + q.close()