diff --git a/.codex/TASK_GROUPBRIEF_FAILURE_CLOSURE.md b/.codex/TASK_GROUPBRIEF_FAILURE_CLOSURE.md new file mode 100644 index 0000000..db58932 --- /dev/null +++ b/.codex/TASK_GROUPBRIEF_FAILURE_CLOSURE.md @@ -0,0 +1,59 @@ +# GroupBrief 故障闭环、周报启用与 Codex 自动维修 + +## 背景 + +9 月 2 日至 5 日出现过图片合同、事实校验、状态汇总和微信 UI 发送失败。图片修复已合入，但周报仍未正式启用，恢复后的群级状态与调度汇总可能不一致，也不存在安全的 Codex 自动维修控制器。 + +## 目标 + +- 补齐日报各阶段稳定错误类型和恢复后的状态重算。 +- 图片来源证明缺失时禁止发送；微信 UI 确定性失败首次即人工锁。 +- 周一日报只生成归档，微信改发上一自然周周报。 +- 周报单群隔离、损坏可见、发送前校验工件哈希，并支持周一漏跑恢复。 +- 建立默认关闭的独立维修队列与控制器，只允许在隔离工作树中调用 Codex、测试并创建 PR。 +- 提供只读 API、健康摘要和归档页状态展示。 + +## 允许修改范围 + +- `app/config`、`app/pipeline`、`app/image`、`app/scheduler`、`app/weekly`、`app/repair` +- `app/api/v2_weekly.py`、`app/api/v2_repair.py`、V2 健康聚合入口 +- `frontend/src/api.ts`、归档页、设置页健康标签及其样式/测试 +- `.env.example`、相关测试和本任务文件 + +## 禁止修改范围 + +- 真实 `.env`、数据库业务数据、现有 `output` 运行记录。 +- 真实 Codex、微信或邮件调用；生产重启、部署、自动合并或自动补发。 +- `.codemap/*`、`PROJECT_REAUDIT.md` 和任何 secrets/认证文件。 +- 自动清除 `UNKNOWN`、目标歧义、数据库损坏或已提交外部操作的人工锁。 + +## 已确定实现要求 + +- `error_type`、阶段、错误指纹和可重试属性必须落到权威状态。 +- 周一替代模式只有周报生成、发送和替代三个开关同时开启时生效。 +- 周报每群独立；损坏 JSON、非法数字、工件写入/读取、哈希不符均形成明确状态。 +- 维修事件输入脱敏；同指纹 7 天一次 PR、全局单任务、每天 2 次、单次 60 分钟、连续 3 次失败熔断 24 小时。 +- `SEND_RESULT_UNKNOWN`、`PROMPT_RESULT_UNKNOWN`、提交结果未知、目标不唯一和损坏状态仅诊断。 +- Codex 固定 `gpt-5.6-sol`，使用 `workspace-write`、自动审批、JSONL、输出 Schema、ephemeral 和隔离工作树。 +- 控制器只接受预定义测试入口，审计 diff 和敏感路径后才允许提交、普通 Push、创建 PR；永不合并或部署。 + +## 验收标准 + +- 新增单元/集成测试全部使用假 Codex、假 Git/GH、临时目录，不触发外部操作。 +- 周一所有日报自动发送旁路均关闭，周二至周日行为不变。 +- 周报损坏和维修状态在 API/归档页/健康摘要中可见。 +- 默认配置不启动维修或周报，不改变当前生产状态。 +- 后端、前端、E2E 与 CI 通过；PR 保持未合并。 + +## 测试命令 + +```powershell +python -m pytest tests -q +cd frontend +npm test -- --run +npm run build +``` + +## 返回格式 + +报告错误闭环、周报行为、维修安全边界、测试证据、Git/PR/CI，以及明确未执行的生产步骤。 diff --git a/.env.example b/.env.example index b0c54dd..55b8d98 100644 --- a/.env.example +++ b/.env.example @@ -125,5 +125,18 @@ RELIABILITY_WATCHDOG_INTERVAL_MINUTES=10 # 周报能力先部署后灰度。14 天验收完成前保持关闭；周报发送另有独立闸门。 WEEKLY_INSIGHTS_ENABLED=false WEEKLY_SEND_ENABLED=false +# 开启后，周一微信发送上一自然周周报，周一日报只生成留档；需与上面两个开关同时开启。 +WEEKLY_REPLACES_MONDAY_DAILY_SEND=false WEEKLY_GENERATE_TIME=07:45 WEEKLY_SEND_TIME=08:30 + +# Codex 自动维修控制器：独立进程，默认关闭；只在隔离 worktree 创建修复 PR。 +REPAIR_ENABLED=false +REPAIR_CODEX_BINARY=codex +REPAIR_WORKTREE_ROOT= +REPAIR_MAX_PER_DAY=2 +REPAIR_TIMEOUT_MINUTES=60 +REPAIR_FINGERPRINT_COOLDOWN_DAYS=7 +REPAIR_CIRCUIT_FAILURE_THRESHOLD=3 +REPAIR_CIRCUIT_COOLDOWN_HOURS=24 +REPAIR_POLL_INTERVAL_MINUTES=10 diff --git a/app/api/v2_repair.py b/app/api/v2_repair.py new file mode 100644 index 0000000..a1e6fac --- /dev/null +++ b/app/api/v2_repair.py @@ -0,0 +1,29 @@ +"""自动维修只读 API。""" + +from fastapi import APIRouter, Depends, HTTPException + +from app.config.settings import Settings, get_settings +from app.repair.store import RepairIncidentStore, public_incident + +router = APIRouter(prefix="/repair", tags=["v2-repair"]) + + +@router.get("/incidents") +def list_repair_incidents(settings: Settings = Depends(get_settings)): + store = RepairIncidentStore(settings) + return { + "schema_version": 1, + "summary": store.summary(), + "items": [public_incident(item) for item in store.list_incidents()], + } + + +@router.get("/incidents/{incident_id}") +def repair_incident_detail( + incident_id: str, + settings: Settings = Depends(get_settings), +): + value = RepairIncidentStore(settings).get(incident_id) + if not value: + raise HTTPException(status_code=404, detail="维修事件不存在") + return public_incident(value) diff --git a/app/api/v2_ui.py b/app/api/v2_ui.py index f4064de..fce5827 100644 --- a/app/api/v2_ui.py +++ b/app/api/v2_ui.py @@ -42,6 +42,7 @@ from app.config.settings import Settings, get_settings from app.v2.constants import RUN_STATE_CORRUPT from app.api.v2_recovery import router as recovery_router +from app.api.v2_repair import router as repair_router from app.api.v2_weekly import router as weekly_router @@ -49,6 +50,7 @@ router.include_router(read_router) router.include_router(image_router) router.include_router(recovery_router) +router.include_router(repair_router) router.include_router(weekly_router) @@ -56,6 +58,47 @@ def system_health(settings: Settings = Depends(get_settings)): checks: dict[str, dict] = {} + from app.repair.store import RepairIncidentStore + + repair = RepairIncidentStore(settings).summary() + checks["auto_repair"] = { + "ok": True, + "status": ( + "CIRCUIT_OPEN" if repair["circuit_open"] else "ENABLED" + if repair["enabled"] else "DISABLED" + ), + "detail": ( + f"queue={repair['queued']} active={repair['active_fingerprint'][:12] or 'none'}" + ), + **repair, + } + + try: + from app.scheduler.manager import get_scheduler + + scheduler = get_scheduler() + job_ids = {job.id for job in scheduler.get_jobs()} if scheduler else set() + except Exception: + job_ids = set() + weekly_generate_registered = "weekly_insights_generate" in job_ids + weekly_send_registered = ( + "daily_wechat_send_batch" in job_ids + if settings.weekly_monday_replacement_enabled + else "weekly_insights_send" in job_ids + ) + checks["weekly_insights"] = { + "ok": not settings.weekly_insights_enabled or weekly_generate_registered, + "status": "ENABLED" if settings.weekly_insights_enabled else "DISABLED", + "detail": ( + f"generate_job={weekly_generate_registered} send_job={weekly_send_registered} " + f"generate_at={settings.weekly_generate_time} send_at={settings.weekly_send_time}" + ), + "generation_enabled": settings.weekly_insights_enabled, + "send_enabled": settings.weekly_send_enabled, + "generation_job_registered": weekly_generate_registered, + "send_job_registered": weekly_send_registered, + } + from app.data_sources.wechat_data_analysis import WeChatDataAnalysisSource source = WeChatDataAnalysisSource(settings=settings) diff --git a/app/api/v2_weekly.py b/app/api/v2_weekly.py index 0c993a5..fdfaffe 100644 --- a/app/api/v2_weekly.py +++ b/app/api/v2_weekly.py @@ -2,6 +2,8 @@ from __future__ import annotations +from datetime import datetime, timedelta +from zoneinfo import ZoneInfo from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import FileResponse @@ -23,10 +25,53 @@ def _public_state(state: dict) -> dict: } +def _next_weekly_at(settings: Settings, clock: str, now: datetime) -> str: + try: + hour, minute = (int(part) for part in clock.split(":", 1)) + except (TypeError, ValueError): + return "" + days = (7 - now.weekday()) % 7 + candidate = (now + timedelta(days=days)).replace( + hour=hour, minute=minute, second=0, microsecond=0 + ) + if candidate <= now: + candidate += timedelta(days=7) + return candidate.isoformat() + + @router.get("") def list_weekly_insights(settings: Settings = Depends(get_settings)): states = [_public_state(item) for item in _store(settings).list_states()] - return {"schema_version": 1, "items": states} + now = datetime.now(ZoneInfo(settings.app_timezone)) + try: + from app.scheduler.manager import get_scheduler + + scheduler = get_scheduler() + job_ids = {job.id for job in scheduler.get_jobs()} if scheduler else set() + except Exception: + job_ids = set() + counts: dict[str, int] = {} + for item in states: + status = str(item.get("status") or "unknown") + counts[status] = counts.get(status, 0) + 1 + return { + "schema_version": 2, + "feature": { + "generation_enabled": settings.weekly_insights_enabled, + "send_enabled": settings.weekly_send_enabled, + "replaces_monday_daily_send": settings.weekly_monday_replacement_enabled, + "next_generate_at": _next_weekly_at(settings, settings.weekly_generate_time, now), + "next_send_at": _next_weekly_at(settings, settings.weekly_send_time, now), + "generation_job_registered": "weekly_insights_generate" in job_ids, + "send_job_registered": ( + "daily_wechat_send_batch" in job_ids + if settings.weekly_monday_replacement_enabled + else "weekly_insights_send" in job_ids + ), + "status_counts": counts, + }, + "items": states, + } @router.get("/{week_start}/{group_id}") diff --git a/app/config/settings.py b/app/config/settings.py index 3eb072f..9ce6a5b 100644 --- a/app/config/settings.py +++ b/app/config/settings.py @@ -3,6 +3,7 @@ from __future__ import annotations from functools import lru_cache +import os from pathlib import Path from typing import Any, Literal @@ -30,8 +31,18 @@ "scheduler_heartbeat_stale_seconds", "weekly_insights_enabled", "weekly_send_enabled", + "weekly_replaces_monday_daily_send", "weekly_generate_time", "weekly_send_time", + "repair_enabled", + "repair_codex_binary", + "repair_worktree_root", + "repair_max_per_day", + "repair_timeout_minutes", + "repair_fingerprint_cooldown_days", + "repair_circuit_failure_threshold", + "repair_circuit_cooldown_hours", + "repair_poll_interval_minutes", "output_root_override", } ) @@ -180,8 +191,20 @@ class Settings(BaseSettings): # 周报能力先部署、后灰度：14 天可靠性验收完成前保持关闭。 weekly_insights_enabled: bool = False weekly_send_enabled: bool = False + # 三个开关同时开启时，周一微信只发送上一自然周周报；日报仍生成留档。 + weekly_replaces_monday_daily_send: bool = False weekly_generate_time: str = "07:45" weekly_send_time: str = "08:30" + # 独立维修进程默认关闭；正式启用前必须先完成 PR/部署验收。 + repair_enabled: bool = False + repair_codex_binary: str = "codex" + repair_worktree_root: str = "" + repair_max_per_day: int = 2 + repair_timeout_minutes: int = 60 + repair_fingerprint_cooldown_days: int = 7 + repair_circuit_failure_threshold: int = 3 + repair_circuit_cooldown_hours: int = 24 + repair_poll_interval_minutes: int = 10 scheduler_heartbeat_stale_seconds: int = 300 # 只供测试/离线执行通过环境变量隔离 output 与相邻 runtime； @@ -199,6 +222,24 @@ def output_dir(self) -> Path: return Path(self.output_root_override).expanduser().resolve() return PROJECT_ROOT / "output" + @property + def weekly_monday_replacement_enabled(self) -> bool: + """周一周报替代日报发送的有效开关，避免半配置时漏发日报。""" + return bool( + self.weekly_insights_enabled + and self.weekly_send_enabled + and self.weekly_replaces_monday_daily_send + ) + + @property + def repair_worktrees_dir(self) -> Path: + if self.repair_worktree_root: + return Path(self.repair_worktree_root).expanduser().resolve() + local_app_data = Path( + os.environ.get("LOCALAPPDATA") or PROJECT_ROOT.parent + ) + return (local_app_data / "GroupBrief" / "repair-worktrees").resolve() + @property def logs_dir(self) -> Path: return PROJECT_ROOT / "logs" diff --git a/app/image/delivery_guard.py b/app/image/delivery_guard.py index 1a69d20..6655d09 100644 --- a/app/image/delivery_guard.py +++ b/app/image/delivery_guard.py @@ -5,6 +5,13 @@ from typing import Any, Mapping +def image_provenance_complete(metadata: Mapping[str, Any] | None) -> bool: + metadata = metadata if isinstance(metadata, Mapping) else {} + if metadata.get("image_enabled") is False: + return True + return all(key in metadata for key in ("image_fallback_level", "image_variant", "image_status")) + + def image_fallback_level(metadata: Mapping[str, Any] | None) -> int: """读取兜底等级；非空脏值按 Level 3 处理，保持 fail-closed。""" metadata = metadata if isinstance(metadata, Mapping) else {} @@ -18,6 +25,10 @@ def image_fallback_level(metadata: Mapping[str, Any] | None) -> int: def image_delivery_eligible(metadata: Mapping[str, Any] | None) -> bool: """只有真实或安全化生成图可以发送；Level 3/Pillow 仅供诊断。""" metadata = metadata if isinstance(metadata, Mapping) else {} + if metadata.get("image_enabled") is False: + return True + if not image_provenance_complete(metadata): + return False fallback_level = image_fallback_level(metadata) image_variant = str(metadata.get("image_variant") or "").strip().lower() image_status = str(metadata.get("image_status") or "").strip().lower() @@ -30,7 +41,8 @@ def image_delivery_eligible(metadata: Mapping[str, Any] | None) -> bool: if ( fallback_level >= 3 or image_variant == "pillow" - or image_status in {"failed", "diagnostic_fallback"} + or image_variant not in {"normal", "safe"} + or image_status not in {"success", "regenerated"} or job_status in {"failed", "ambiguous_result", "diagnostic_fallback"} ): return False diff --git a/app/image/regeneration.py b/app/image/regeneration.py index 14fa991..6104635 100644 --- a/app/image/regeneration.py +++ b/app/image/regeneration.py @@ -283,6 +283,16 @@ def _promote_image( image_sha256=_sha256(target), **success_fields, ) + from app.scheduler.daily_v2_job import ( + ScheduleStateCorruptionError, + reconcile_daily_schedule_from_runs, + ) + + settings = Settings(_env_file=None, output_root_override=str(store.root)) + try: + reconcile_daily_schedule_from_runs(settings, run_date) + except ScheduleStateCorruptionError: + logger.warning("图片已恢复，但调度状态损坏，保留人工复核：run_date=%s", run_date) return store.load_run(group_name, run_date) diff --git a/app/pipeline/daily_pipeline.py b/app/pipeline/daily_pipeline.py index fdbdd0b..fb87c62 100644 --- a/app/pipeline/daily_pipeline.py +++ b/app/pipeline/daily_pipeline.py @@ -605,6 +605,15 @@ def send_due_for_dates( """ now = now or datetime.now(ZoneInfo(self.settings.app_timezone)) normalized_dates = sorted({validate_run_date(value) for value in run_dates}) + if ( + not recovery + and self.settings.weekly_monday_replacement_enabled + and now.weekday() == 0 + ): + # 防御性门禁：即使绕过 APScheduler 直接调用 send-due， + # 本周一日报也只留档，不进入微信自动发送。 + current_date = now.date().isoformat() + normalized_dates = [value for value in normalized_dates if value != current_date] results: list[dict] = [] groups = self._load_groups() due_group_ids: list[int] = [] @@ -757,6 +766,9 @@ def _write_runtime_status_safe(self, run_dates: list[str]) -> None: for run_date in run_dates: try: write_daily_status(self.store, run_date) + from app.repair.events import capture_daily_incidents + + capture_daily_incidents(self.settings, self.store, run_date) except Exception: logger.exception("每日运行报告写入失败：run_date=%s", run_date) diff --git a/app/pipeline/delivery_stages.py b/app/pipeline/delivery_stages.py index b7b3f31..a845b88 100644 --- a/app/pipeline/delivery_stages.py +++ b/app/pipeline/delivery_stages.py @@ -10,7 +10,7 @@ from app.config.settings import Settings from app.core.observability import log_event from app.db.models import Group -from app.image.delivery_guard import image_delivery_eligible +from app.image.delivery_guard import image_delivery_eligible, image_provenance_complete from app.image.image_task import verify_image from app.pipeline.stage_result import StageResult from app.sender.base import WechatSender @@ -19,8 +19,13 @@ FAILED, IMAGE_FALLBACK_NOT_SENDABLE, IMAGE_FILE_MISSING, + IMAGE_PROVENANCE_MISSING, READY_TO_SEND, SENT, + WECHAT_IMAGE_NOT_STAGED, + WECHAT_TARGET_AMBIGUOUS, + WECHAT_TARGET_NOT_FOUND, + WECHAT_TEXT_NOT_STAGED, ) from app.v2.run_store import RunStore @@ -64,6 +69,64 @@ def __init__( self._name_sync_audit = name_sync_audit self.logger = logger + @staticmethod + def _manual_pre_submit_error(detail: str, stage: str) -> str: + text = str(detail or "") + if "匹配数 0" in text or "当前 0" in text or "未得到可验证的目标匹配" in text: + return WECHAT_TARGET_NOT_FOUND + if "歧义" in text or "数量不是 1" in text: + return WECHAT_TARGET_AMBIGUOUS + if stage == "text" and "未观察到输入区暂存" in text: + return WECHAT_TEXT_NOT_STAGED + if stage == "image" and "未观察到预览" in text: + return WECHAT_IMAGE_NOT_STAGED + return "" + + def _hold_manual_pre_submit_failure( + self, + context: DeliveryContext, + *, + stage: str, + error_type: str, + detail: str, + finished_at: str, + diagnostics: dict, + ) -> dict: + persisted, _ = self.store.finish_send_claim( + context.group_name, + context.run_date, + context.claim_id, + send_state="failed_final", + status=context.run.get("status", READY_TO_SEND), + send_hold=True, + send_hold_reason=error_type, + needs_manual_send=True, + send_next_retry_at="", + send_retry_attempt_count=int(context.run.get("send_retry_attempt_count") or 0), + send_error=str(detail)[:500], + send_error_type=error_type, + **{ + f"{stage}_attempt_finished_at": finished_at, + f"{stage}_submitted_at": "", + f"{stage}_verification_diagnostics": diagnostics, + }, + ) + if not persisted: + return self.finish_unknown( + context.group_name, + context.run_date, + context.claim_id, + stage, + f"人工锁状态无法持久化：{detail}", + diagnostics=diagnostics, + ) + return { + "group_name": context.group_name, + "status": "held", + "error_type": error_type, + "detail": detail, + } + def run( self, group: Group, @@ -125,13 +188,17 @@ def _claim(self, context: DeliveryContext) -> StageResult[DeliveryContext]: allow_sent=context.allow_sent, ) if not claim_id: - if claim_reason == IMAGE_FALLBACK_NOT_SENDABLE: + if claim_reason in {IMAGE_FALLBACK_NOT_SENDABLE, IMAGE_PROVENANCE_MISSING}: return StageResult.stop( { "group_name": context.group_name, "status": "failed", - "error_type": IMAGE_FALLBACK_NOT_SENDABLE, - "detail": "Level 3/Pillow 诊断图不可发送", + "error_type": claim_reason, + "detail": ( + "图片来源元数据不完整，不可发送" + if claim_reason == IMAGE_PROVENANCE_MISSING + else "Level 3/Pillow 诊断图不可发送" + ), } ) if claim_reason == "result_unknown": @@ -177,27 +244,37 @@ def _prepare_payload( context: DeliveryContext, ) -> StageResult[DeliveryContext]: if not image_delivery_eligible(context.run): - detail = "Level 3/Pillow 诊断图不可发送，已在发送前预检阶段拦截" + provenance_missing = not image_provenance_complete(context.run) + error_type = ( + IMAGE_PROVENANCE_MISSING + if provenance_missing + else IMAGE_FALLBACK_NOT_SENDABLE + ) + detail = ( + "图片来源元数据不完整，已在发送前预检阶段拦截" + if provenance_missing + else "Level 3/Pillow 诊断图不可发送，已在发送前预检阶段拦截" + ) self.store.finish_send_claim( context.group_name, context.run_date, context.claim_id, send_state="held", send_hold=True, - send_hold_reason=IMAGE_FALLBACK_NOT_SENDABLE, + send_hold_reason=error_type, needs_manual_send=False, status=FAILED, failed_stage="image", error=detail, - error_type=IMAGE_FALLBACK_NOT_SENDABLE, + error_type=error_type, send_error=detail, - send_error_type=IMAGE_FALLBACK_NOT_SENDABLE, + send_error_type=error_type, ) return StageResult.stop( { "group_name": context.group_name, "status": "failed", - "error_type": IMAGE_FALLBACK_NOT_SENDABLE, + "error_type": error_type, "detail": detail, } ) @@ -388,6 +465,18 @@ def _send_text( diagnostics=result.diagnostics, ) ) + manual_error = self._manual_pre_submit_error(result.detail, "text") + if manual_error: + return StageResult.stop( + self._hold_manual_pre_submit_failure( + context, + stage="text", + error_type=manual_error, + detail=result.detail, + finished_at=finished_at, + diagnostics=result.diagnostics, + ) + ) persisted, failed_run, final = self.store.finish_send_failure( context.group_name, context.run_date, @@ -527,6 +616,18 @@ def _send_image( diagnostics=result.diagnostics, ) ) + manual_error = self._manual_pre_submit_error(result.detail, "image") + if manual_error: + return StageResult.stop( + self._hold_manual_pre_submit_failure( + context, + stage="image", + error_type=manual_error, + detail=result.detail, + finished_at=finished_at, + diagnostics=result.diagnostics, + ) + ) persisted, failed_run, final = self.store.finish_send_failure( context.group_name, context.run_date, diff --git a/app/pipeline/generation_stages.py b/app/pipeline/generation_stages.py index 107fa6e..a7df7cc 100644 --- a/app/pipeline/generation_stages.py +++ b/app/pipeline/generation_stages.py @@ -364,6 +364,7 @@ def _load_or_fetch_messages( status=FAILED, failed_stage="data", error="群未绑定微信群 ID", + error_type=WECHAT_DATA_UNAVAILABLE, ) return StageResult.stop( { @@ -542,6 +543,17 @@ def _refresh_snapshot_and_ranking( ranking, template_name=context.group.ranking_template, ) + attribution = build_attribution_contract(messages) + snapshot_path = self.store.messages_path(context.group_name, context.run_date) + self._save_json(snapshot_path, [message.to_dict() for message in messages]) + self._save_json( + self.store.ranking_json_path(context.group_name, context.run_date), + ranking.to_dict(), + ) + self.store.ranking_txt_path(context.group_name, context.run_date).write_text( + ranking_txt, + encoding="utf-8", + ) except Exception as exc: context.timings["ranking_ms"] = round((perf_counter() - started_at) * 1000) self.store.update( @@ -552,6 +564,7 @@ def _refresh_snapshot_and_ranking( error=context.run.get("error"), message_refresh_status="failed", message_refresh_error=str(exc)[:300], + message_refresh_error_type=RANKING_FAILED, ) return StageResult.stop( { @@ -563,17 +576,6 @@ def _refresh_snapshot_and_ranking( ) context.timings["ranking_ms"] = round((perf_counter() - started_at) * 1000) - attribution = build_attribution_contract(messages) - snapshot_path = self.store.messages_path(context.group_name, context.run_date) - self._save_json(snapshot_path, [message.to_dict() for message in messages]) - self._save_json( - self.store.ranking_json_path(context.group_name, context.run_date), - ranking.to_dict(), - ) - self.store.ranking_txt_path(context.group_name, context.run_date).write_text( - ranking_txt, - encoding="utf-8", - ) next_status = SENT if context.run.get("status") == SENT else RANKING_READY prior_hold_reason = str(context.run.get("send_hold_reason") or "") hold_reason = ( @@ -637,6 +639,18 @@ def _build_ranking( ), name_source=getattr(context.group, "sender_name_policy", "resolved"), ) + self._save_json( + self.store.ranking_json_path(context.group_name, context.run_date), + ranking.to_dict(), + ) + ranking_txt = self.renderer.render( + ranking, + template_name=context.group.ranking_template, + ) + self.store.ranking_txt_path(context.group_name, context.run_date).write_text( + ranking_txt, + encoding="utf-8", + ) except Exception as exc: self.store.update( context.group_name, @@ -644,6 +658,7 @@ def _build_ranking( status=FAILED, failed_stage="ranking", error=str(exc)[:300], + error_type=RANKING_FAILED, ) context.timings["ranking_ms"] = round((perf_counter() - started_at) * 1000) return StageResult.stop( @@ -655,18 +670,6 @@ def _build_ranking( ) context.timings["ranking_ms"] = round((perf_counter() - started_at) * 1000) - self._save_json( - self.store.ranking_json_path(context.group_name, context.run_date), - ranking.to_dict(), - ) - ranking_txt = self.renderer.render( - ranking, - template_name=context.group.ranking_template, - ) - self.store.ranking_txt_path(context.group_name, context.run_date).write_text( - ranking_txt, - encoding="utf-8", - ) self.store.update( context.group_name, context.run_date, diff --git a/app/pipeline/image_stages.py b/app/pipeline/image_stages.py index 5b882a7..a4084d7 100644 --- a/app/pipeline/image_stages.py +++ b/app/pipeline/image_stages.py @@ -7,7 +7,7 @@ from typing import Callable import uuid -from app.image.delivery_guard import image_delivery_eligible +from app.image.delivery_guard import image_delivery_eligible, image_provenance_complete from app.image.image_task import ImageJob, SerialImageQueue from app.image.regeneration import normalize_candidate_diagnostics from app.core.logging import get_logger @@ -15,6 +15,7 @@ from app.v2.constants import ( FAILED, IMAGE_FALLBACK_NOT_SENDABLE, + IMAGE_PROVENANCE_MISSING, IMAGE_GENERATION_FAILED, IMAGE_READY, READY_TO_SEND, @@ -133,8 +134,12 @@ def record_result(self, job: ImageJob, result: dict) -> None: image_metadata = { "image_fallback_level": generator_detail.get("fallback_level"), "image_variant": generator_detail.get("image_variant"), + "image_status": result.get("status"), } - diagnostic_fallback = not image_delivery_eligible(image_metadata) + diagnostic_fallback = bool( + int(generator_detail.get("fallback_level") or 0) >= 3 + or str(generator_detail.get("image_variant") or "").lower() == "pillow" + ) image_size_bytes = ( job.output_path.stat().st_size if (result["success"] or diagnostic_fallback) and job.output_path.is_file() @@ -183,7 +188,7 @@ def record_result(self, job: ImageJob, result: dict) -> None: failed_stage="image" if not result["success"] else None, error=error_detail, image_error=error_detail, - image_status=result["status"], + image_status="success" if result["success"] else result["status"], error_type=error_type if not result["success"] else None, stage_timings=stage_timings, imagegen_ms=imagegen_ms, @@ -263,14 +268,25 @@ def advance_ready(self, job: ImageJob, run_date: str) -> None: run = self.store.load_run(job.group_name, run_date) if run.get("status") == IMAGE_READY: if not image_delivery_eligible(run): + provenance_missing = not image_provenance_complete(run) + error_type = ( + IMAGE_PROVENANCE_MISSING + if provenance_missing + else IMAGE_FALLBACK_NOT_SENDABLE + ) + detail = ( + "图片来源元数据不完整，不可进入发送流程" + if provenance_missing + else "Level 3/Pillow 诊断图不可进入发送流程" + ) self.store.update( job.group_name, run_date, status=FAILED, failed_stage="image", - error="Level 3/Pillow 诊断图不可进入发送流程", - image_error="Level 3/Pillow 诊断图不可进入发送流程", - error_type=IMAGE_FALLBACK_NOT_SENDABLE, + error=detail, + image_error=detail, + error_type=error_type, ) return self.store.update(job.group_name, run_date, status=READY_TO_SEND) diff --git a/app/repair/__init__.py b/app/repair/__init__.py new file mode 100644 index 0000000..e502a25 --- /dev/null +++ b/app/repair/__init__.py @@ -0,0 +1,5 @@ +"""GroupBrief 脱敏故障队列与独立 Codex 维修控制器。""" + +from app.repair.store import RepairIncidentStore + +__all__ = ["RepairIncidentStore"] diff --git a/app/repair/codex_result.schema.json b/app/repair/codex_result.schema.json new file mode 100644 index 0000000..0101e01 --- /dev/null +++ b/app/repair/codex_result.schema.json @@ -0,0 +1,16 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "summary": {"type": "string"}, + "root_cause": {"type": "string"}, + "regression_test_added": {"type": "boolean"}, + "regression_test_reproduced": {"type": "boolean"}, + "changed_files": {"type": "array", "items": {"type": "string"}}, + "tests_run": {"type": "array", "items": {"type": "string"}}, + "tests_passed": {"type": "boolean"}, + "safe_to_propose_pr": {"type": "boolean"} + }, + "required": ["summary", "root_cause", "regression_test_added", "regression_test_reproduced", "changed_files", "tests_run", "tests_passed", "safe_to_propose_pr"], + "additionalProperties": false +} diff --git a/app/repair/controller.py b/app/repair/controller.py new file mode 100644 index 0000000..e50631a --- /dev/null +++ b/app/repair/controller.py @@ -0,0 +1,287 @@ +"""独立 Codex 维修控制器；只在隔离 worktree 产出待确认 PR。""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Callable + +from app.config.settings import PROJECT_ROOT, Settings +from app.repair.store import RepairIncidentStore, sanitize_text +from app.v2.run_store import _run_mutex + +_ALLOWED_PREFIXES = ("app/", "tests/", "frontend/", "scripts/", "docs/", ".env.example") +_SENSITIVE_PARTS = ( + ".env", "auth.json", "cookie", "token", "credential", "secret", "browser-data", +) +_TESTS_BY_SCOPE = { + "ranking": ["tests/test_v2_pipeline.py"], + "data": ["tests/test_v2_pipeline.py", "tests/test_v2_data_source.py"], + "prompt": ["tests/test_v2_prompt_builder.py", "tests/test_v2_pipeline.py"], + "image": ["tests/test_v2_image_task.py", "tests/test_v2_pipeline.py"], + "weekly": ["tests/test_weekly_insights.py"], + "scheduler": ["tests/test_scheduler.py", "tests/test_reliability_watchdog.py"], +} + + +@dataclass(frozen=True) +class CommandResult: + returncode: int + stdout: str = "" + stderr: str = "" + + +class CommandRunner: + def run( + self, + argv: list[str], + *, + cwd: Path, + stdin: str = "", + timeout: int = 300, + ) -> CommandResult: + environment = os.environ.copy() + executable = Path(argv[0]).name.lower() if argv else "" + if executable not in {"git", "git.exe", "gh", "gh.exe"}: + for key in tuple(environment): + upper = key.upper() + if upper in {"OPENAI_API_KEY", "CODEX_API_KEY"} or any( + marker in upper for marker in ("COOKIE", "TOKEN", "PASSWORD") + ): + environment.pop(key, None) + completed = subprocess.run( + argv, + cwd=str(cwd), + input=stdin or None, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout, + shell=False, + env=environment, + ) + return CommandResult( + completed.returncode, + completed.stdout[-20000:], + completed.stderr[-12000:], + ) + + +class RepairController: + def __init__( + self, + settings: Settings, + *, + store: RepairIncidentStore | None = None, + runner: CommandRunner | None = None, + repository_root: Path | None = None, + ) -> None: + self.settings = settings + self.store = store or RepairIncidentStore(settings) + self.runner = runner or CommandRunner() + self.repository_root = (repository_root or PROJECT_ROOT).resolve() + + def _run(self, argv: list[str], *, cwd: Path, stdin: str = "", timeout: int = 300) -> CommandResult: + result = self.runner.run(argv, cwd=cwd, stdin=stdin, timeout=timeout) + if result.returncode != 0: + detail = sanitize_text(result.stderr or result.stdout, limit=800) + raise RuntimeError(f"命令失败({argv[0]} {argv[1] if len(argv) > 1 else ''})：{detail}") + return result + + @staticmethod + def _default_branch(symbolic: str) -> str: + match = re.search(r"refs/remotes/origin/([^\s]+)", symbolic) + if not match: + raise RuntimeError("无法解析 origin/HEAD 默认分支") + return match.group(1) + + def _prompt(self, incident: dict, tests: list[str]) -> str: + payload = { + "fingerprint": incident["fingerprint"], + "scope": incident["scope"], + "error_type": incident["error_type"], + "stage": incident["stage"], + "source_path": incident.get("source_path", ""), + "error_summary": incident.get("redacted_error_summary", ""), + "related_commit_sha": incident.get("related_commit_sha", ""), + "required_tests": tests, + } + return ( + "你在隔离 Git worktree 中修复 GroupBrief 的一个确定性代码故障。" + "只能修改 app/、tests/、frontend/、scripts/、docs/ 或 .env.example；" + "必须先新增能复现问题的回归测试，再做最小修复并运行指定测试。" + "结构化结果必须明确回归测试在修复前已复现、修复后已通过。" + "禁止读取 .env、认证文件、原始群聊、Cookie、Token、微信截图；" + "禁止发送消息、部署、重启、提交、Push、创建或合并 PR。" + "最终严格按输出 Schema 返回。\n故障：" + + json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + ) + + @staticmethod + def _changed_paths(porcelain: str) -> list[str]: + paths: list[str] = [] + for line in porcelain.splitlines(): + if len(line) < 4: + continue + value = line[3:].strip().replace("\\", "/") + if " -> " in value: + value = value.split(" -> ", 1)[1] + paths.append(value.strip('"')) + return sorted(set(paths)) + + @staticmethod + def _audit_paths(paths: list[str]) -> None: + if not paths: + raise RuntimeError("Codex 未产生任何修改") + for path in paths: + lowered = path.lower() + if ".." in Path(path).parts or any(part in lowered for part in _SENSITIVE_PARTS): + raise RuntimeError(f"检测到敏感或越界文件：{path}") + if not ( + path == ".env.example" + or any(path.startswith(prefix) for prefix in _ALLOWED_PREFIXES[:-1]) + ): + raise RuntimeError(f"检测到范围外修改：{path}") + if not any(path.startswith("tests/") or path.endswith(".test.ts") or "/e2e/" in path for path in paths): + raise RuntimeError("未新增或修改回归测试") + + @staticmethod + def _audit_diff(diff: str) -> None: + added = "\n".join( + line[1:] for line in diff.splitlines() + if line.startswith("+") and not line.startswith("+++") + ) + secret_patterns = ( + r"sk-[A-Za-z0-9_-]{20,}", + r"(?i)bearer\s+[A-Za-z0-9._~+/=-]{20,}", + r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----", + r"(?i)(?:api[_-]?key|password|cookie|token)\s*[:=]\s*['\"][^'\"]{12,}['\"]", + ) + if any(re.search(pattern, added) for pattern in secret_patterns): + raise RuntimeError("diff 中疑似包含敏感数据") + + def run_once(self) -> dict: + if not self.settings.repair_enabled: + return {"status": "disabled"} + with _run_mutex(self.store.root / "controller-process.lock"): + incident, reason = self.store.start_next() + if incident is None: + return {"status": "not_run", "reason": reason} + try: + return self._execute(incident) + except subprocess.TimeoutExpired: + timeout_minutes = max(min(int(self.settings.repair_timeout_minutes), 60), 1) + return self.store.finish( + incident, + success=False, + reason=f"维修任务超过 {timeout_minutes} 分钟", + ) + except Exception as exc: + return self.store.finish(incident, success=False, reason=str(exc)) + + def _execute(self, incident: dict) -> dict: + tests = _TESTS_BY_SCOPE.get(str(incident.get("scope") or ""), ["tests/test_scheduler.py"]) + self._run(["git", "fetch", "--prune", "origin"], cwd=self.repository_root) + symbolic = self._run( + ["git", "symbolic-ref", "refs/remotes/origin/HEAD"], + cwd=self.repository_root, + ).stdout + base = self._default_branch(symbolic) + branch = f"codex/fix-auto-{incident['fingerprint'][:10]}-{incident['incident_id'][:6]}" + worktree = (self.settings.repair_worktrees_dir / incident["incident_id"]).resolve() + root = self.settings.repair_worktrees_dir.resolve() + if root not in worktree.parents or worktree.exists(): + raise RuntimeError("维修 worktree 路径不安全或已存在") + root.mkdir(parents=True, exist_ok=True) + self._run( + ["git", "worktree", "add", "-b", branch, str(worktree), f"origin/{base}"], + cwd=self.repository_root, + ) + incident["branch"] = branch + self.store.save(incident) + + result_path = worktree / ".codex-repair-result.json" + schema_path = worktree / "app" / "repair" / "codex_result.schema.json" + command = [ + self.settings.repair_codex_binary, + "exec", + "--model", "gpt-5.6-sol", + "--sandbox", "workspace-write", + "--approve-for-me", + "--json", + "--output-schema", str(schema_path), + "--output-last-message", str(result_path), + "--ephemeral", + "--ignore-user-config", + "--ignore-rules", + "--cd", str(worktree), + "-", + ] + codex = self._run( + command, + cwd=worktree, + stdin=self._prompt(incident, tests), + timeout=max(min(int(self.settings.repair_timeout_minutes), 60), 1) * 60, + ) + thread_match = re.search(r'"type"\s*:\s*"thread.started".*?"thread_id"\s*:\s*"([^"]+)"', codex.stdout) + incident["codex_thread_id"] = thread_match.group(1) if thread_match else "" + try: + result = json.loads(result_path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise RuntimeError("Codex 结构化结果缺失或无效") from exc + finally: + if result_path.exists(): + result_path.unlink() + if not all( + bool(result.get(key)) + for key in ( + "regression_test_added", + "regression_test_reproduced", + "tests_passed", + "safe_to_propose_pr", + ) + ): + raise RuntimeError("Codex 未满足回归测试与安全 PR 条件") + + status = self._run(["git", "status", "--porcelain"], cwd=worktree).stdout + paths = self._changed_paths(status) + self._audit_paths(paths) + self._run(["git", "diff", "--check"], cwd=worktree) + diff = self._run(["git", "diff", "--no-ext-diff", "--unified=0"], cwd=worktree).stdout + self._audit_diff(diff) + test_command = [sys.executable, "-m", "pytest", *tests, "-q"] + test_result = self._run(test_command, cwd=worktree, timeout=1800) + incident["test_result"] = { + "command": "python -m pytest " + " ".join(tests) + " -q", + "passed": True, + "summary": sanitize_text(test_result.stdout, limit=500), + } + self.store.save(incident) + + self._run(["git", "add", "--", *paths], cwd=worktree) + self._run(["git", "diff", "--cached", "--check"], cwd=worktree) + self._run(["git", "commit", "-m", f"fix: 自动修复 {incident['error_type'].lower()}"], cwd=worktree) + commit_sha = self._run(["git", "rev-parse", "HEAD"], cwd=worktree).stdout.strip() + incident["commit_sha"] = commit_sha + self.store.save(incident) + self._run(["git", "push", "-u", "origin", branch], cwd=worktree) + pr = self._run( + [ + "gh", "pr", "create", "--base", base, "--head", branch, + "--title", f"fix: 自动修复 {incident['error_type']}", + "--body", ( + f"自动维修事件 `{incident['incident_id']}`。\n\n" + f"指纹：`{incident['fingerprint']}`\n\n" + "仅创建 PR，未合并、部署、重启或执行任何外部发送。" + ), + ], + cwd=worktree, + ) + incident["pr_url"] = pr.stdout.strip().splitlines()[-1] + return self.store.finish(incident, success=True) diff --git a/app/repair/events.py b/app/repair/events.py new file mode 100644 index 0000000..21e4314 --- /dev/null +++ b/app/repair/events.py @@ -0,0 +1,94 @@ +"""从权威脱敏状态生成 RepairIncident；不读取消息、Prompt 或认证文件。""" + +from __future__ import annotations + +import hashlib +from pathlib import Path + +from app.config.settings import Settings +from app.repair.store import RepairIncidentStore +from app.scheduler.daily_v2_job import DailyScheduleState +from app.v2.run_store import RunStore +from app.weekly.store import WeeklyStore + + +def _opaque_group(run: dict) -> str: + value = str(run.get("group_id") or run.get("group_name") or "unknown") + return hashlib.sha256(value.encode("utf-8")).hexdigest()[:12] + + +def capture_daily_incidents( + settings: Settings, + run_store: RunStore, + run_date: str, +) -> list[dict]: + incidents: list[dict] = [] + ledger = RepairIncidentStore(settings) + for run in run_store.list_runs(run_date): + error_type = str(run.get("error_type") or run.get("send_error_type") or "") + if not error_type: + continue + stage = str(run.get("failed_stage") or run.get("stage") or "unknown").lower() + scope = "send" if stage == "send" or error_type.startswith("SEND_") else stage + incidents.append( + ledger.record( + scope=scope, + error_type=error_type, + stage=stage, + source_path=f"daily/{run_date}/group-{_opaque_group(run)}/run.json", + error_summary=str(run.get("error") or run.get("send_error") or ""), + ) + ) + scheduler_state = DailyScheduleState(settings.output_dir).load(run_date) + if scheduler_state.get("state_status") == "corrupt": + incidents.append( + ledger.record( + scope="scheduler", + error_type="SCHEDULER_STATE_CORRUPT", + stage="state", + source_path=f".scheduler/{run_date}.json", + error_summary=str(scheduler_state.get("state_error_reason") or ""), + ) + ) + return incidents + + +def capture_weekly_incidents(settings: Settings) -> list[dict]: + incidents: list[dict] = [] + ledger = RepairIncidentStore(settings) + for state in WeeklyStore(settings.output_dir).list_states(): + error_type = str(state.get("error_type") or "") + if not error_type: + continue + group_id = int(state.get("group_id") or 0) + source = ( + f".weekly/{state.get('week_start', '')}_{state.get('week_end', '')}/" + f"group-{group_id}/weekly.json" + ) + incidents.append( + ledger.record( + scope="weekly", + error_type=error_type, + stage=str(state.get("stage") or "weekly"), + source_path=source, + error_summary=str(state.get("error_summary") or state.get("send_error") or ""), + ) + ) + return incidents + + +def capture_persisted_incidents(settings: Settings) -> list[dict]: + store = RunStore(settings.output_dir) + dates = sorted( + { + str(run.get("run_date") or "") + for run in store.list_runs() + if str(run.get("run_date") or "") + }, + reverse=True, + )[:2] + incidents: list[dict] = [] + for run_date in dates: + incidents.extend(capture_daily_incidents(settings, store, run_date)) + incidents.extend(capture_weekly_incidents(settings)) + return incidents diff --git a/app/repair/store.py b/app/repair/store.py new file mode 100644 index 0000000..f97987b --- /dev/null +++ b/app/repair/store.py @@ -0,0 +1,274 @@ +"""文件型 RepairIncident 账本：跨进程原子、脱敏、去重与熔断。""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import uuid +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any + +from app.config.settings import Settings +from app.v2.run_store import _run_mutex + +_SECRET = re.compile( + r"(?i)(api[_-]?key|token|password|cookie|authorization)\s*[:=]\s*[^\s,;]+" +) +_BEARER = re.compile(r"(?i)bearer\s+[a-z0-9._~+/=-]+") +_UNKNOWN_OR_DANGEROUS = { + "SEND_RESULT_UNKNOWN", + "PROMPT_RESULT_UNKNOWN", + "WEEKLY_SEND_RESULT_UNKNOWN", + "WEEKLY_AI_RESULT_UNKNOWN", + "SCHEDULER_STATE_CORRUPT", + "RUN_STATE_CORRUPT", + "WEEKLY_STATE_CORRUPT", + "GROUP_TARGET_MISMATCH", + "WECHAT_TARGET_AMBIGUOUS", + "WECHAT_TARGET_NOT_FOUND", +} +_ENVIRONMENT_ERRORS = { + "WEEKLY_AI_FAILED_FALLBACK", + "WECHAT_OFFLINE", + "WECHAT_DATA_UNAVAILABLE", +} + + +def sanitize_text(value: object, *, limit: int = 500) -> str: + text = str(value or "").replace("\x00", " ") + text = _SECRET.sub(r"\1=[REDACTED]", text) + text = _BEARER.sub("Bearer [REDACTED]", text) + return " ".join(text.split())[:limit] + + +def repair_mode_for(scope: str, error_type: str) -> str: + code = str(error_type or "").upper() + if scope == "repair": + return "diagnostic_only" + if code in _UNKNOWN_OR_DANGEROUS or "UNKNOWN" in code or "CORRUPT" in code: + return "diagnostic_only" + if code in _ENVIRONMENT_ERRORS: + return "environment" + if scope in {"send", "wechat"}: + return "environment" + return "code_fix" + + +class RepairIncidentStore: + def __init__(self, settings: Settings): + self.settings = settings + self.root = Path(settings.output_dir) / ".repair" + self.incidents_dir = self.root / "incidents" + self.state_path = self.root / "controller.json" + self.lock_path = self.root / "admission.lock" + + @staticmethod + def _now(now: datetime | None = None) -> datetime: + return now or datetime.now().astimezone() + + def _write(self, path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temp = path.with_suffix(path.suffix + ".tmp") + temp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + os.replace(temp, path) + + def _read(self, path: Path) -> dict: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return {} + return value if isinstance(value, dict) else {} + + def list_incidents(self) -> list[dict]: + if not self.incidents_dir.is_dir(): + return [] + items: list[dict] = [] + for path in self.incidents_dir.glob("*.json"): + value = self._read(path) + if value: + items.append(value) + else: + items.append( + { + "incident_id": path.stem, + "status": "corrupt", + "repair_mode": "diagnostic_only", + "error_type": "REPAIR_INCIDENT_CORRUPT", + } + ) + items.sort(key=lambda item: str(item.get("created_at") or ""), reverse=True) + return items + + def get(self, incident_id: str) -> dict: + if not re.fullmatch(r"[a-f0-9]{32}", str(incident_id or "")): + return {} + return self._read(self.incidents_dir / f"{incident_id}.json") + + def save(self, incident: dict) -> dict: + incident_id = str(incident.get("incident_id") or "") + if not re.fullmatch(r"[a-f0-9]{32}", incident_id): + raise ValueError("incident_id 无效") + incident = dict(incident) + incident["updated_at"] = self._now().isoformat() + self._write(self.incidents_dir / f"{incident_id}.json", incident) + return incident + + def record( + self, + *, + scope: str, + error_type: str, + stage: str, + source_path: str, + error_summary: str = "", + stack_summary: str = "", + related_commit_sha: str = "", + now: datetime | None = None, + ) -> dict: + now = self._now(now) + safe_scope = sanitize_text(scope, limit=40).lower() or "unknown" + safe_error = sanitize_text(error_type, limit=80).upper() or "UNKNOWN_ERROR" + safe_stage = sanitize_text(stage, limit=40).lower() or "unknown" + safe_source = sanitize_text(source_path, limit=180).replace("\\", "/") + if ":/" in safe_source or safe_source.startswith("/"): + safe_source = Path(safe_source).name + summary = sanitize_text(error_summary) + stack = sanitize_text(stack_summary) + canonical = "|".join( + (safe_scope, safe_error, safe_stage, stack or summary, related_commit_sha[:40]) + ) + fingerprint = hashlib.sha256(canonical.encode("utf-8")).hexdigest() + cutoff = now - timedelta(days=max(int(self.settings.repair_fingerprint_cooldown_days), 1)) + with _run_mutex(self.lock_path): + for existing in self.list_incidents(): + if existing.get("fingerprint") != fingerprint: + continue + try: + created = datetime.fromisoformat(str(existing.get("created_at") or "")) + if created.tzinfo is None: + created = created.replace(tzinfo=now.tzinfo) + except ValueError: + continue + if created >= cutoff: + existing["last_seen_at"] = now.isoformat() + existing["occurrence_count"] = int(existing.get("occurrence_count") or 1) + 1 + return self.save(existing) + incident_id = uuid.uuid4().hex + mode = repair_mode_for(safe_scope, safe_error) + incident = { + "schema_version": 1, + "incident_id": incident_id, + "fingerprint": fingerprint, + "scope": safe_scope, + "error_type": safe_error, + "stage": safe_stage, + "source_path": safe_source, + "related_commit_sha": sanitize_text(related_commit_sha, limit=40), + "redacted_error_summary": summary, + "redacted_stack_summary": stack, + "repair_mode": mode, + "status": "queued" if mode == "code_fix" else mode, + "attempt_count": 0, + "created_at": now.isoformat(), + "updated_at": now.isoformat(), + "last_seen_at": now.isoformat(), + "occurrence_count": 1, + "cooldown_until": "", + "codex_thread_id": "", + "branch": "", + "commit_sha": "", + "pr_url": "", + "test_result": {}, + "circuit_breaker_reason": "", + } + self._write(self.incidents_dir / f"{incident_id}.json", incident) + return incident + + def controller_state(self) -> dict: + return self._read(self.state_path) + + def start_next(self, *, now: datetime | None = None) -> tuple[dict | None, str]: + now = self._now(now) + with _run_mutex(self.lock_path): + state = self.controller_state() + try: + circuit_until = datetime.fromisoformat(str(state.get("circuit_until") or "")) + except (TypeError, ValueError): + circuit_until = None + if circuit_until and circuit_until > now: + return None, "circuit_open" + today = now.date().isoformat() + attempts = [value for value in state.get("attempts", []) if str(value).startswith(today)] + if len(attempts) >= max(int(self.settings.repair_max_per_day), 1): + return None, "daily_limit" + queued = [item for item in self.list_incidents() if item.get("status") == "queued"] + if not queued: + return None, "empty" + incident = sorted(queued, key=lambda item: str(item.get("created_at") or ""))[0] + incident.update( + status="running", + attempt_count=int(incident.get("attempt_count") or 0) + 1, + last_attempt_at=now.isoformat(), + ) + self.save(incident) + state["active_incident_id"] = incident["incident_id"] + state["active_fingerprint"] = incident["fingerprint"] + state["attempts"] = (list(state.get("attempts") or []) + [now.isoformat()])[-30:] + self._write(self.state_path, state) + return incident, "started" + + def finish(self, incident: dict, *, success: bool, reason: str = "", now: datetime | None = None) -> dict: + now = self._now(now) + with _run_mutex(self.lock_path): + state = self.controller_state() + streak = 0 if success else int(state.get("failure_streak") or 0) + 1 + state.update(active_incident_id="", active_fingerprint="", failure_streak=streak) + if not success and streak >= max(int(self.settings.repair_circuit_failure_threshold), 1): + until = now + timedelta(hours=max(int(self.settings.repair_circuit_cooldown_hours), 1)) + state["circuit_until"] = until.isoformat() + state["circuit_reason"] = sanitize_text(reason) + incident["circuit_breaker_reason"] = state["circuit_reason"] + incident["cooldown_until"] = until.isoformat() + self._write(self.state_path, state) + incident["status"] = "pr_created" if success else "failed" + if reason: + incident["last_error"] = sanitize_text(reason) + return self.save(incident) + + def summary(self) -> dict: + items = self.list_incidents() + state = self.controller_state() + now = self._now() + circuit_until = str(state.get("circuit_until") or "") + try: + circuit_open = datetime.fromisoformat(circuit_until) > now + except (TypeError, ValueError): + circuit_open = False + counts: dict[str, int] = {} + for item in items: + status = str(item.get("status") or "unknown") + counts[status] = counts.get(status, 0) + 1 + return { + "enabled": bool(self.settings.repair_enabled), + "queued": sum(item.get("status") == "queued" for item in items), + "active_fingerprint": str(state.get("active_fingerprint") or ""), + "circuit_open": circuit_open, + "circuit_until": circuit_until, + "circuit_reason": str(state.get("circuit_reason") or ""), + "incident_count": len(items), + "status_counts": counts, + } + + +def public_incident(value: dict) -> dict: + allowed = { + "schema_version", "incident_id", "fingerprint", "scope", "error_type", + "stage", "source_path", "repair_mode", "status", "attempt_count", + "created_at", "updated_at", "last_seen_at", "occurrence_count", + "cooldown_until", "codex_thread_id", "branch", "commit_sha", "pr_url", + "test_result", "circuit_breaker_reason", "last_error", + } + return {key: value.get(key) for key in allowed if key in value} diff --git a/app/scheduler/daily_v2_job.py b/app/scheduler/daily_v2_job.py index 2f252cc..2e9cce5 100644 --- a/app/scheduler/daily_v2_job.py +++ b/app/scheduler/daily_v2_job.py @@ -22,8 +22,16 @@ from app.pipeline.daily_pipeline import DailyPipeline, parse_date from app.services.generation_runtime import GenerationBusyError, generation_mutex from app.services.email_service import email_delivery_config_error -from app.v2.constants import IMAGE_GENERATION_FAILED, SCHEDULER_STATE_CORRUPT -from app.v2.run_store import _atomic_write_text, _run_mutex +from app.v2.constants import ( + CORRUPT, + FAILED, + IMAGE_GENERATION_FAILED, + IMAGE_READY, + READY_TO_SEND, + SCHEDULER_STATE_CORRUPT, + SENT, +) +from app.v2.run_store import RunStore, _atomic_write_text, _run_mutex from app.scheduler.outcome import ProcessExitCode, attach_outcome, summarize_results from app.scheduler.task_manifest import ( build_expected_groups, @@ -244,12 +252,31 @@ def run_daily_v2_job( } except Exception as exc: logger.exception("V2 每日任务异常") - result = {"status": "failed", "detail": str(exc)[:300]} + result = { + "status": "failed", + "error_type": "UNEXPECTED_SCHEDULER_ERROR", + "stage": "scheduler", + "retryable": True, + "detail": str(exc)[:300], + } return _finalize_invocation(settings, parsed_date.isoformat(), result) def _finalize_invocation(settings: Settings, run_date: str, result: dict) -> dict: finalized = attach_outcome(result) + if finalized.get("error_type") == "UNEXPECTED_SCHEDULER_ERROR": + try: + from app.repair.store import RepairIncidentStore + + RepairIncidentStore(settings).record( + scope="scheduler", + error_type="UNEXPECTED_SCHEDULER_ERROR", + stage="scheduler", + source_path=f".scheduler/{run_date}.json", + error_summary=str(finalized.get("detail") or ""), + ) + except Exception: + logger.exception("调度异常写入维修队列失败：run_date=%s", run_date) logger.info( "V2 每日任务终态：run_date=%s source_status=%s outcome=%s exit_code=%d", run_date, @@ -326,6 +353,9 @@ def ensure_daily_manifest( parsed, timezone=settings.app_timezone, schedule_send_time=settings.schedule_send_time, + weekly_replaces_monday_daily_send=( + settings.weekly_monday_replacement_enabled + ), resolver=resolver, ) manifest = manifest_fields(expected) @@ -684,6 +714,93 @@ def _generation_status(results: list[dict]) -> str: return str(summarize_results(results)["outcome_status"]) +def reconcile_daily_schedule_from_runs(settings: Settings, run_date: str) -> dict: + """以任务清单和权威 run.json 全量重算生成汇总，不触碰发送结果。""" + + state_store = DailyScheduleState(settings.output_dir) + state = state_store.load(run_date) + if state.get("state_status") == "corrupt": + raise ScheduleStateCorruptionError("调度状态文件损坏，禁止自动覆盖") + expected = state.get("expected_groups") + if not isinstance(expected, list): + return state + + runs_by_group_id: dict[int, dict] = {} + runs_by_name: dict[str, dict] = {} + for run in RunStore(settings.output_dir).list_runs(run_date): + try: + group_id = int(run.get("group_id") or 0) + except (TypeError, ValueError): + group_id = 0 + if group_id > 0: + runs_by_group_id[group_id] = run + name = str(run.get("group_name") or "") + if name: + runs_by_name[name] = run + + results: list[dict] = [] + recovered_groups: list[str] = [] + for item in expected: + group_id = int(item["group_id"]) + group_name = str(item.get("group_name") or group_id) + run = runs_by_group_id.get(group_id) or runs_by_name.get(group_name) + if not run: + results.append( + { + "group_name": group_name, + "status": "failed_final", + "error_type": "RUN_STATE_MISSING", + "failed_stage": "reconcile", + "detail": "预期群缺少权威 run.json", + } + ) + continue + status = str(run.get("status") or "") + if status == SENT: + result_status = "success" + elif status in {IMAGE_READY, READY_TO_SEND}: + result_status = "ready_to_send" + elif status == CORRUPT: + result_status = "blocked" + elif status == FAILED: + result_status = ( + "held" if bool(run.get("send_hold") or run.get("needs_manual_send")) + else "failed_final" + ) + else: + result_status = "failed_final" + result = { + "group_name": group_name, + "status": result_status, + "error_type": run.get("error_type"), + "failed_stage": run.get("failed_stage"), + "detail": run.get("error") or run.get("image_error") or "", + "recovery_status": run.get("image_recovery_status"), + "recovered_at": run.get("image_recovered_at") or run.get("image_regenerated_at"), + "receipt_source": run.get("image_receipt_source"), + "codex_thread_id": run.get("codex_thread_id"), + } + results.append(result) + if result_status in {"success", "ready_to_send"} and result.get("recovered_at"): + recovered_groups.append(group_name) + + compact = _compact_results(results) + next_status = _generation_status(results) + fields: dict = { + "generation_results": compact, + "generation_status": next_status, + "generation_reconciled_at": _now_iso(), + "generation_recovery_groups": sorted(set(recovered_groups)), + "generation_hold": any( + item.get("status") in {"held", "blocked", "failed_final"} + for item in results + ), + } + if _generation_results_terminal(results): + fields["generation_completed_at"] = state.get("generation_completed_at") or _now_iso() + return state_store.update(run_date, **fields) + + def _generation_results_terminal(results: list[dict]) -> bool: """批次内所有群都已成功或进入明确人工/最终终态时才封存批次。""" if not results: diff --git a/app/scheduler/manager.py b/app/scheduler/manager.py index ffeab55..f9751fa 100644 --- a/app/scheduler/manager.py +++ b/app/scheduler/manager.py @@ -3,13 +3,16 @@ from __future__ import annotations from datetime import datetime, time, timedelta +import subprocess +import sys from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.triggers.date import DateTrigger +from apscheduler.triggers.interval import IntervalTrigger from zoneinfo import ZoneInfo -from app.config.settings import Settings, get_settings +from app.config.settings import PROJECT_ROOT, Settings, get_settings from app.core.logging import get_logger from app.scheduler.daily_v2_job import DailyScheduleState, run_daily_v2_job from app.scheduler.heartbeat import record_scheduler_heartbeat @@ -18,7 +21,8 @@ from app.scheduler.send_job import run_send_due_job from app.v2.constants import EXECUTION_WAIT_RETRY, IMAGE_READY, READY_TO_SEND from app.v2.run_store import RunStore -from app.weekly.service import WeeklyInsightsService +from app.weekly.service import WeeklyInsightsService, previous_natural_week +from app.weekly.store import WeeklyStore logger = get_logger("groupbrief.scheduler") @@ -76,6 +80,19 @@ def _parse_weekly_time(value: str) -> time: ) +def _is_current_monday_weekly_replacement( + settings: Settings, + now: datetime, + run_date: str, +) -> bool: + """仅替换本周一当天的日报发送，历史日期仍保持人工恢复边界。""" + return bool( + settings.weekly_monday_replacement_enabled + and now.weekday() == 0 + and run_date == now.date().isoformat() + ) + + def _timestamp(value: object, *, now: datetime) -> datetime | None: if not isinstance(value, str) or not value.strip(): return None @@ -162,6 +179,10 @@ def _schedule_on_demand_jobs( if today not in selected_dates: return scheduled + if _is_current_monday_weekly_replacement(settings, now, today): + # 日报仍生成并保留 READY_TO_SEND，但周一微信入口由周报接管。 + return scheduled + send_clock = _parse_send_time(settings.schedule_send_time) due_at = datetime.combine(now.date(), send_clock, tzinfo=now.tzinfo) if now < due_at: @@ -257,6 +278,8 @@ def run_scheduled_send_batch(run_date: str | None = None) -> dict: now = _normalize_now(settings) target_date = run_date or now.date().isoformat() try: + if _is_current_monday_weekly_replacement(settings, now, target_date): + return run_scheduled_weekly_send(settings=settings, now=now) return run_send_due_job( settings=settings, now=now, @@ -294,6 +317,7 @@ def run_scheduled_startup_recovery() -> dict: run_dates=recovery_dates(now, settings.reliability_lookback_days), include_newly_ready_send=False, ) + _schedule_weekly_replacement_jobs(_scheduler, settings, now=now) record_scheduler_heartbeat( settings, job="startup_recovery", @@ -302,28 +326,135 @@ def run_scheduled_startup_recovery() -> dict: return result -def run_scheduled_weekly_insights() -> dict: +def run_scheduled_weekly_insights( + *, + settings: Settings | None = None, + now: datetime | None = None, +) -> dict: """周一独立生成上一自然周归档；不读取原始聊天、不发送。""" - settings = get_settings() + settings = settings or get_settings() + supplied_now = now + now = _normalize_now(settings, now) record_scheduler_heartbeat(settings, job="weekly_insights", status="started") - result = WeeklyInsightsService(settings).generate_previous_week() + try: + result = WeeklyInsightsService(settings).generate_previous_week(now=now) + except Exception as exc: + record_scheduler_heartbeat( + settings, + job="weekly_insights", + status="error", + detail=f"{type(exc).__name__}: {exc}", + ) + raise record_scheduler_heartbeat( settings, job="weekly_insights", status=str(result.get("status") or "unknown"), ) + _schedule_weekly_replacement_jobs( + _scheduler, + settings, + now=(now if supplied_now is not None else _normalize_now(settings)), + include_generation=False, + ) return result -def run_scheduled_weekly_send() -> dict: +def run_scheduled_weekly_send( + *, + settings: Settings | None = None, + now: datetime | None = None, +) -> dict: """可选周报发送独立运行，不再参与日报批次或空闲日志。""" - settings = get_settings() - results = WeeklyInsightsService(settings).send_due() + settings = settings or get_settings() + now = _normalize_now(settings, now) + results = WeeklyInsightsService(settings).send_due(now=now) outcome = summarize_results(results) require_scheduler_success(outcome, allow_not_run=True) return outcome +def run_repair_worker_process() -> dict: + """在独立进程消费脱敏维修队列；Web 服务自身不修改代码工作树。""" + settings = get_settings() + completed = subprocess.run( + [sys.executable, str(PROJECT_ROOT / "scripts" / "repair_worker.py")], + cwd=str(PROJECT_ROOT), + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=max(min(int(settings.repair_timeout_minutes), 60), 1) * 60 + 300, + shell=False, + ) + if completed.returncode != 0: + raise RuntimeError((completed.stderr or completed.stdout)[-500:]) + return {"status": "complete", "detail": completed.stdout[-500:]} + + +def _schedule_weekly_replacement_jobs( + scheduler: BackgroundScheduler | None, + settings: Settings, + *, + now: datetime | None = None, + include_generation: bool = True, +) -> list[str]: + """周一重启或延迟生成后，按周报持久化状态恢复一次性任务。""" + if scheduler is None: + return [] + now = _normalize_now(settings, now) + today = now.date().isoformat() + if not _is_current_monday_weekly_replacement(settings, now, today): + return [] + + scheduled: list[str] = [] + generate_due = datetime.combine( + now.date(), + _parse_weekly_time(settings.weekly_generate_time), + tzinfo=now.tzinfo, + ) + if include_generation and now >= generate_due: + job_id = f"weekly_generate_once_{today.replace('-', '')}" + _add_one_shot( + scheduler, + job_id=job_id, + name=f"WeeklyGenerateOnce:{today}", + func=run_scheduled_weekly_insights, + run_at=now + timedelta(seconds=1), + ) + scheduled.append(job_id) + + send_due = datetime.combine( + now.date(), + _parse_clock( + settings.weekly_send_time, + fallback=_DEFAULT_SEND_TIME, + field_name="weekly_send_time", + ), + tzinfo=now.tzinfo, + ) + if now < send_due: + return scheduled + week_start, week_end = previous_natural_week(now.date()) + sendable = any( + state.get("week_start") == week_start.isoformat() + and state.get("week_end") == week_end.isoformat() + and state.get("status") in {"ready_to_send", "sending"} + for state in WeeklyStore(settings.output_dir).list_states() + ) + if sendable: + job_id = f"weekly_send_once_{today.replace('-', '')}" + _add_one_shot( + scheduler, + job_id=job_id, + name=f"WeeklySendOnce:{today}", + func=run_scheduled_weekly_send, + run_at=now + timedelta(seconds=1), + ) + scheduled.append(job_id) + return scheduled + + def _schedule_startup_recovery( scheduler: BackgroundScheduler, settings: Settings, @@ -396,7 +527,7 @@ def start_scheduler(settings: Settings) -> BackgroundScheduler: coalesce=True, max_instances=1, ) - if settings.weekly_send_enabled: + if settings.weekly_send_enabled and not settings.weekly_monday_replacement_enabled: weekly_send_time = _parse_clock( settings.weekly_send_time, fallback=_DEFAULT_SEND_TIME, @@ -417,10 +548,24 @@ def start_scheduler(settings: Settings) -> BackgroundScheduler: coalesce=True, max_instances=1, ) + if settings.repair_enabled: + scheduler.add_job( + run_repair_worker_process, + trigger=IntervalTrigger( + minutes=max(int(settings.repair_poll_interval_minutes), 1), + timezone=tz, + ), + id="repair_controller", + name="RepairController", + misfire_grace_time=300, + coalesce=True, + max_instances=1, + ) scheduler.start() _scheduler = scheduler record_scheduler_heartbeat(settings, job="scheduler", status="started") _schedule_startup_recovery(scheduler, settings) + _schedule_weekly_replacement_jobs(scheduler, settings) logger.info( "调度已启动：每日 %s 生成，%s 微信串行发送批次（时区 %s）", generate_time.strftime("%H:%M"), diff --git a/app/scheduler/recovery_planner.py b/app/scheduler/recovery_planner.py index 02b3e37..732caa3 100644 --- a/app/scheduler/recovery_planner.py +++ b/app/scheduler/recovery_planner.py @@ -113,6 +113,9 @@ def preview( datetime.fromisoformat(run_date).date(), timezone=self.settings.app_timezone, schedule_send_time=self.settings.schedule_send_time, + weekly_replaces_monday_daily_send=( + self.settings.weekly_monday_replacement_enabled + ), ) source = "current_config_preview" runs = { @@ -206,6 +209,9 @@ def confirm_generation( datetime.fromisoformat(run_date).date(), timezone=self.settings.app_timezone, schedule_send_time=self.settings.schedule_send_time, + weekly_replaces_monday_daily_send=( + self.settings.weekly_monday_replacement_enabled + ), resolver=pipeline.period_resolver, ) state = self.state_store.load(run_date) @@ -270,6 +276,9 @@ def repair_empty_manifest_and_generate( datetime.fromisoformat(run_date).date(), timezone=self.settings.app_timezone, schedule_send_time=self.settings.schedule_send_time, + weekly_replaces_monday_daily_send=( + self.settings.weekly_monday_replacement_enabled + ), resolver=pipeline.period_resolver, ) actual_ids = sorted( diff --git a/app/scheduler/reliability_watchdog.py b/app/scheduler/reliability_watchdog.py index bd5b964..559a7c1 100644 --- a/app/scheduler/reliability_watchdog.py +++ b/app/scheduler/reliability_watchdog.py @@ -104,20 +104,29 @@ def run_reliability_watchdog( } generation_results.append(result) - try: - # 历史任务只能生成恢复；微信自动发送仍只扫描当天。 - send_results = DailyPipeline(settings=settings).send_due_for_dates( - [now.date().isoformat()], now=now, recovery=False - ) - except Exception as exc: - logger.exception("启动恢复发送检查异常") + if settings.weekly_monday_replacement_enabled and now.weekday() == 0: + # 周一日报只生成留档；周报补偿由 scheduler manager 的独立任务负责。 send_results = [ { - "status": "failed", - "error_type": type(exc).__name__, - "detail": str(exc)[:300], + "status": "not_run", + "detail": "周一日报微信发送已由上一自然周周报替代", } ] + else: + try: + # 历史任务只能生成恢复；微信自动发送仍只扫描当天。 + send_results = DailyPipeline(settings=settings).send_due_for_dates( + [now.date().isoformat()], now=now, recovery=False + ) + except Exception as exc: + logger.exception("启动恢复发送检查异常") + send_results = [ + { + "status": "failed", + "error_type": type(exc).__name__, + "detail": str(exc)[:300], + } + ] status = "success" if any(item.get("status") in {"failed", "partial"} for item in generation_results + send_results): diff --git a/app/scheduler/runtime_status.py b/app/scheduler/runtime_status.py index 829b0a1..57c702f 100644 --- a/app/scheduler/runtime_status.py +++ b/app/scheduler/runtime_status.py @@ -171,7 +171,13 @@ def _group_node_status( ) -> str: if node_id == "scheduler": return "success" if scheduler_started else "pending" - if node_id == "image" and not image_delivery_eligible(run): + image_job = run.get("image_job") if isinstance(run.get("image_job"), dict) else {} + image_attempt_finished = bool( + run.get("image_status") + or image_job.get("status") in {"completed", "failed", "ambiguous_result", "diagnostic_fallback"} + or str(run.get("status") or "") in {"IMAGE_READY", "READY_TO_SEND", "SENT"} + ) + if node_id == "image" and image_attempt_finished and not image_delivery_eligible(run): # 历史 SENT 仍保持发送成功，但图片节点必须呈现诊断失败事实。 return "failed" diff --git a/app/scheduler/task_manifest.py b/app/scheduler/task_manifest.py index 44bce85..e5d399b 100644 --- a/app/scheduler/task_manifest.py +++ b/app/scheduler/task_manifest.py @@ -21,10 +21,12 @@ def build_expected_groups( *, timezone: str, schedule_send_time: str = "08:30", + weekly_replaces_monday_daily_send: bool = False, resolver: PeriodResolver | None = None, ) -> list[dict]: resolver = resolver or PeriodResolver() expected: list[dict] = [] + monday_replaced = bool(weekly_replaces_monday_daily_send and run_date.weekday() == 0) for group in groups: if group.id is None: continue @@ -65,8 +67,13 @@ def build_expected_groups( "image_prompt_override": str(group.image_prompt_override or ""), "send_target": str(group.send_target or ""), "wechat_send_enabled": bool(group.wechat_send_enabled), + "wechat_send_replaced_by_weekly": bool( + monday_replaced and group.wechat_send_enabled + ), "expected_terminal": ( - "SENT" if group.wechat_send_enabled else "READY_TO_SEND" + "SENT" + if group.wechat_send_enabled and not monday_replaced + else "READY_TO_SEND" ), "period_start": window.period_start.isoformat(), "period_end": window.period_end.isoformat(), diff --git a/app/v2/constants.py b/app/v2/constants.py index 9275b4e..7faca19 100644 --- a/app/v2/constants.py +++ b/app/v2/constants.py @@ -65,7 +65,12 @@ IMAGE_FILE_MISSING = "IMAGE_FILE_MISSING" IMAGE_CONTENT_VERIFICATION_FAILED = "IMAGE_CONTENT_VERIFICATION_FAILED" IMAGE_FALLBACK_NOT_SENDABLE = "IMAGE_FALLBACK_NOT_SENDABLE" +IMAGE_PROVENANCE_MISSING = "IMAGE_PROVENANCE_MISSING" WECHAT_OFFLINE = "WECHAT_OFFLINE" +WECHAT_TARGET_NOT_FOUND = "WECHAT_TARGET_NOT_FOUND" +WECHAT_TARGET_AMBIGUOUS = "WECHAT_TARGET_AMBIGUOUS" +WECHAT_TEXT_NOT_STAGED = "WECHAT_TEXT_NOT_STAGED" +WECHAT_IMAGE_NOT_STAGED = "WECHAT_IMAGE_NOT_STAGED" SEND_TEXT_FAILED = "SEND_TEXT_FAILED" SEND_IMAGE_FAILED = "SEND_IMAGE_FAILED" RUN_STATE_CORRUPT = "RUN_STATE_CORRUPT" diff --git a/app/v2/reliability.py b/app/v2/reliability.py index b6b0fb8..6d3e0e0 100644 --- a/app/v2/reliability.py +++ b/app/v2/reliability.py @@ -34,6 +34,10 @@ READY_TO_SEND, SENT, WECHAT_DATA_UNAVAILABLE, + WECHAT_IMAGE_NOT_STAGED, + WECHAT_TARGET_AMBIGUOUS, + WECHAT_TARGET_NOT_FOUND, + WECHAT_TEXT_NOT_STAGED, ) CHECKPOINT_BY_STATUS = { @@ -86,6 +90,11 @@ "MESSAGE_SNAPSHOT_INVALID", "MISSED_SEND_WINDOW", "GROUP_TARGET_MISMATCH", + WECHAT_TARGET_NOT_FOUND, + WECHAT_TARGET_AMBIGUOUS, + WECHAT_TEXT_NOT_STAGED, + WECHAT_IMAGE_NOT_STAGED, + "IMAGE_PROVENANCE_MISSING", IMAGE_FALLBACK_NOT_SENDABLE, } ) diff --git a/app/v2/run_store.py b/app/v2/run_store.py index f962ad6..79b69c5 100644 --- a/app/v2/run_store.py +++ b/app/v2/run_store.py @@ -27,7 +27,7 @@ validate_iso_date, validate_path_label, ) -from app.image.delivery_guard import image_delivery_eligible +from app.image.delivery_guard import image_delivery_eligible, image_provenance_complete from app.services.handoff_service import safe_dir_name from app.v2.constants import ( CORRUPT, @@ -41,6 +41,7 @@ FILE_RANKING_TXT, FILE_RUN, IMAGE_FALLBACK_NOT_SENDABLE, + IMAGE_PROVENANCE_MISSING, IMAGE_READY, PENDING, PROMPT_READY, @@ -654,7 +655,11 @@ def claim_send( if data.get("send_hold") and not allow_hold: return None, data, "send_hold" if not image_delivery_eligible(data): - return None, data, IMAGE_FALLBACK_NOT_SENDABLE + return None, data, ( + IMAGE_FALLBACK_NOT_SENDABLE + if image_provenance_complete(data) + else IMAGE_PROVENANCE_MISSING + ) prompt_meta = ( data.get("prompt_meta") if isinstance(data.get("prompt_meta"), dict) diff --git a/app/weekly/service.py b/app/weekly/service.py index 3dab09e..a1032e9 100644 --- a/app/weekly/service.py +++ b/app/weekly/service.py @@ -17,6 +17,7 @@ from app.db import repository as repo from app.db.models import Group from app.image.fallback import _fit_lines, _load_font +from app.image.image_task import verify_image from app.providers.ai.base import ExternalCallResultUnknownError from app.providers.ai.codex import build_summary_provider from app.sender.base import WechatSender @@ -26,6 +27,7 @@ from app.services.group_provider_config import resolve_group_ai_settings from app.v2.run_store import RunStore from app.weekly.store import WeeklyStore +from app.repair.store import RepairIncidentStore def previous_natural_week(reference: date) -> tuple[date, date]: @@ -46,6 +48,29 @@ def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() +def _nonnegative_int(value: object, field: str) -> int: + try: + parsed = int(value or 0) + except (TypeError, ValueError) as exc: + raise ValueError(f"{field} 不是合法整数") from exc + if parsed < 0: + raise ValueError(f"{field} 不能为负数") + return parsed + + +def _failure_fields(error_type: str, stage: str, detail: str) -> dict: + summary = str(detail)[:300] + return { + "error_type": error_type, + "stage": stage, + "error_summary": summary, + "failure_fingerprint": hashlib.sha256( + f"weekly|{error_type}|{stage}|{summary}".encode("utf-8") + ).hexdigest(), + "retryable": False, + } + + class WeeklyInsightsService: def __init__( self, @@ -82,7 +107,53 @@ def generate_previous_week( ) start, end = previous_natural_week(now.date()) groups = self._groups(group_ids) - results = [self._generate_group(group, start, end, now) for group in groups] + results: list[dict] = [] + for group in groups: + try: + results.append(self._generate_group(group, start, end, now)) + except Exception as exc: + assert group.id is not None + error_type = ( + "WEEKLY_DATA_INVALID" + if isinstance(exc, (TypeError, ValueError)) + else "WEEKLY_ARTIFACT_WRITE_FAILED" + ) + detail = str(exc)[:300] + fingerprint = hashlib.sha256( + f"weekly|{error_type}|{detail}".encode("utf-8") + ).hexdigest() + self.store.save( + start.isoformat(), + end.isoformat(), + group.id, + { + "status": "needs_attention", + "group_name": group.display_name, + "error_type": error_type, + "stage": "aggregate" if error_type == "WEEKLY_DATA_INVALID" else "artifact", + "error_summary": detail, + "failure_fingerprint": fingerprint, + "retryable": False, + "generated_at": now.isoformat(), + }, + ) + RepairIncidentStore(self.settings).record( + scope="weekly", + error_type=error_type, + stage="aggregate" if error_type == "WEEKLY_DATA_INVALID" else "artifact", + source_path=f".weekly/{start.isoformat()}_{end.isoformat()}/group-{group.id}/weekly.json", + error_summary=detail, + now=now, + ) + results.append( + { + "group_id": group.id, + "group_name": group.display_name, + "status": "held", + "error_type": error_type, + "detail": detail, + } + ) return { "status": "complete" if all(item["status"] in {"ready_to_send", "skipped"} for item in results) else "partial", "week_start": start.isoformat(), @@ -128,7 +199,7 @@ def _generate_group( end_text, group.id, status="needs_attention", - error_type=error_type, + **_failure_fields(error_type, "external_call", "上次周报外部操作中断"), ) return { "group_id": group.id, @@ -212,12 +283,20 @@ def _generate_group( self._render_card(group, aggregate, start_text, end_text, card_path) text_bytes = narrative.encode("utf-8") card_bytes = card_path.read_bytes() + result_unknown = ai_status == "result_unknown" + error_type = ( + "WEEKLY_AI_RESULT_UNKNOWN" + if result_unknown + else "WEEKLY_AI_FAILED_FALLBACK" + if ai_status == "failed" + else "" + ) payload = self.store.save( start_text, end_text, group.id, { - "status": "ready_to_send", + "status": "needs_attention" if result_unknown else "ready_to_send", "group_name": group.display_name, "wechat_group_id": group.wechat_group_id, "send_target_snapshot": effective_send_target(group), @@ -227,6 +306,13 @@ def _generate_group( "ai_status": ai_status, "ai_error": ai_error, "ai_call_count": ai_call_count, + "error_type": error_type, + "stage": "ai" if error_type else "complete", + "retryable": False, + "failure_fingerprint": ( + hashlib.sha256(f"weekly|{error_type}|{ai_error}".encode("utf-8")).hexdigest() + if error_type else "" + ), "requested_provider": requested_provider, "requested_model": requested_model, "actual_provider": actual_provider, @@ -256,8 +342,14 @@ def _aggregate(self, group: Group, week_start: date, week_end: date) -> dict: ranking = _safe_json(self.daily_store.ranking_json_path(group.display_name, run_date)) if not ranking: missing_days.append(run_date) - message_count = int(ranking.get("message_count") or run.get("message_count") or 0) - speaker_count = int(ranking.get("speaker_count") or run.get("speaker_count") or 0) + message_count = _nonnegative_int( + ranking.get("message_count") or run.get("message_count") or 0, + f"{run_date}.message_count", + ) + speaker_count = _nonnegative_int( + ranking.get("speaker_count") or run.get("speaker_count") or 0, + f"{run_date}.speaker_count", + ) daily.append( { "date": run_date, @@ -274,7 +366,10 @@ def _aggregate(self, group: Group, week_start: date, week_end: date) -> dict: identity = str(row.get("identity_key") or f"name:{name.casefold()}") item = contributors.setdefault(identity, {"identity_key": identity, "name": name, "count": 0}) item["name"] = name - item["count"] += int(row.get("count") or 0) + item["count"] += _nonnegative_int( + row.get("count") or 0, + f"{run_date}.top_speakers.count", + ) prompt_meta = run.get("prompt_meta") if isinstance(run.get("prompt_meta"), dict) else {} selection = prompt_meta.get("topic_selection") if isinstance(prompt_meta.get("topic_selection"), dict) else {} candidates = selection.get("candidates") if isinstance(selection.get("candidates"), list) else [] @@ -393,9 +488,13 @@ def send_due(self, *, now: datetime | None = None) -> list[dict]: week_end.isoformat(), group_id, status="needs_attention", - error_type="WEEKLY_SEND_RESULT_UNKNOWN", send_claim_id="", send_claim_expires_at="", + **_failure_fields( + "WEEKLY_SEND_RESULT_UNKNOWN", + "send_claim", + "发送租约过期，提交结果未知", + ), ) results.append( { @@ -415,10 +514,49 @@ def send_due(self, *, now: datetime | None = None) -> list[dict]: if target != str(state.get("send_target_snapshot") or ""): self.store.update( week_start.isoformat(), week_end.isoformat(), group_id, - status="needs_attention", error_type="WEEKLY_SEND_TARGET_CHANGED", + status="needs_attention", + **_failure_fields( + "WEEKLY_SEND_TARGET_CHANGED", + "send_preflight", + "发送目标与生成时快照不一致", + ), ) results.append({"group_name": group.display_name, "status": "held", "error_type": "WEEKLY_SEND_TARGET_CHANGED"}) continue + text_path = self.store.text_path(week_start.isoformat(), week_end.isoformat(), group_id) + card_path = self.store.card_path(week_start.isoformat(), week_end.isoformat(), group_id) + try: + text_bytes = text_path.read_bytes() + card_bytes = card_path.read_bytes() + except OSError as exc: + error_type = "WEEKLY_ARTIFACT_MISSING" + self.store.update( + week_start.isoformat(), week_end.isoformat(), group_id, + status="needs_attention", + send_error=str(exc)[:300], + **_failure_fields(error_type, "send_preflight", str(exc)), + ) + results.append({"group_name": group.display_name, "status": "held", "error_type": error_type}) + continue + image_ok, image_detail = verify_image(card_path) + hashes_match = ( + _sha256_bytes(text_bytes) == str(state.get("text_sha256") or "") + and _sha256_bytes(card_bytes) == str(state.get("card_sha256") or "") + ) + if not image_ok or not text_bytes.strip() or not hashes_match: + error_type = "WEEKLY_ARTIFACT_HASH_MISMATCH" + self.store.update( + week_start.isoformat(), week_end.isoformat(), group_id, + status="needs_attention", + send_error=(image_detail if not image_ok else "周报文字或卡片哈希不一致"), + **_failure_fields( + error_type, + "send_preflight", + image_detail if not image_ok else "周报文字或卡片哈希不一致", + ), + ) + results.append({"group_name": group.display_name, "status": "held", "error_type": error_type}) + continue claim_id, state = self.store.claim_send( week_start.isoformat(), week_end.isoformat(), @@ -427,24 +565,35 @@ def send_due(self, *, now: datetime | None = None) -> list[dict]: ) if not claim_id: continue - text_path = self.store.text_path(week_start.isoformat(), week_end.isoformat(), group_id) - card_path = self.store.card_path(week_start.isoformat(), week_end.isoformat(), group_id) try: text_result, image_result = self.sender.send_bundle( target, - text_path.read_text(encoding="utf-8"), + text_bytes.decode("utf-8"), card_path, ) except Exception as exc: self.store.update( week_start.isoformat(), week_end.isoformat(), group_id, - status="needs_attention", error_type="WEEKLY_SEND_RESULT_UNKNOWN", + status="needs_attention", send_error=str(exc)[:300], send_claim_id="", send_claim_expires_at="", + **_failure_fields("WEEKLY_SEND_RESULT_UNKNOWN", "send", str(exc)), ) results.append({"group_name": group.display_name, "status": "held", "error_type": "WEEKLY_SEND_RESULT_UNKNOWN"}) break - image_ok = image_result is not None and image_result.success - if text_result.success and image_ok: + text_ok = bool( + text_result.success + and text_result.submitted + and not text_result.outcome_unknown + and text_result.verification_level == "ui_observed" + ) + image_ok = bool( + image_result is not None + and image_result.success + and image_result.submitted + and not image_result.outcome_unknown + and image_result.verification_level == "ui_observed" + ) + if text_ok and image_ok: self.store.update( week_start.isoformat(), week_end.isoformat(), group_id, status="sent", sent_at=now.isoformat(), send_target=target, @@ -459,15 +608,21 @@ def send_due(self, *, now: datetime | None = None) -> list[dict]: continue unknown = bool( text_result.outcome_unknown - or text_result.submitted - or (image_result and (image_result.outcome_unknown or image_result.submitted)) + or (image_result and image_result.outcome_unknown) + or (text_result.submitted and not text_ok) + or (image_result and image_result.submitted and not image_ok) ) error_type = "WEEKLY_SEND_RESULT_UNKNOWN" if unknown else "WEEKLY_SEND_FAILED" self.store.update( week_start.isoformat(), week_end.isoformat(), group_id, - status="needs_attention", error_type=error_type, + status="needs_attention", send_claim_id="", send_claim_expires_at="", send_error=f"text={text_result.detail}; image={getattr(image_result, 'detail', '')}"[:300], + **_failure_fields( + error_type, + "send", + f"text={text_result.detail}; image={getattr(image_result, 'detail', '')}", + ), ) results.append({"group_name": group.display_name, "status": "held", "error_type": error_type}) if unknown: diff --git a/app/weekly/store.py b/app/weekly/store.py index e91a89d..ccc89b0 100644 --- a/app/weekly/store.py +++ b/app/weekly/store.py @@ -43,15 +43,31 @@ def load(self, week_start: str, week_end: str, group_id: int) -> dict: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): - return { - "schema_version": 1, - "week_start": week_start, - "week_end": week_end, - "group_id": group_id, - "status": "needs_attention", - "error_type": "WEEKLY_STATE_CORRUPT", - } - return value if isinstance(value, dict) else {} + return self._corrupt(week_start, week_end, group_id, "read_or_json_invalid") + if not isinstance(value, dict): + return self._corrupt(week_start, week_end, group_id, "root_not_object") + if ( + value.get("week_start") != week_start + or value.get("week_end") != week_end + or value.get("group_id") != group_id + ): + return self._corrupt(week_start, week_end, group_id, "identity_invalid") + return value + + @staticmethod + def _corrupt(week_start: str, week_end: str, group_id: int, reason: str) -> dict: + return { + "schema_version": 1, + "week_start": week_start, + "week_end": week_end, + "group_id": group_id, + "group_name": f"群 {group_id}", + "status": "needs_attention", + "error_type": "WEEKLY_STATE_CORRUPT", + "stage": "state", + "retryable": False, + "state_error_reason": reason, + } def save(self, week_start: str, week_end: str, group_id: int, value: dict) -> dict: with _LOCK: @@ -109,15 +125,18 @@ def list_states(self) -> list[dict]: states: list[dict] = [] for path in self.root.glob("*_*/group-*/weekly.json"): try: - value = json.loads(path.read_text(encoding="utf-8")) - except (OSError, UnicodeError, json.JSONDecodeError): + period = path.parent.parent.name + week_start, week_end = period.split("_", 1) + group_id = int(path.parent.name.removeprefix("group-")) + value = self.load(week_start, week_end, group_id) + except (ValueError, OSError): continue if isinstance(value, dict): states.append(value) states.sort( key=lambda item: ( str(item.get("week_start") or ""), - int(item.get("group_id") or 0), + int(item.get("group_id") or 0) if str(item.get("group_id") or "").isdigit() else 0, ), reverse=True, ) diff --git a/frontend/src/api.ts b/frontend/src/api.ts index dc54f9d..e35139e 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -370,6 +370,43 @@ export interface WeeklyInsight { topics: { title: string; days: number }[]; }; card_url?: string; + error_type?: string; + state_error_reason?: string; +} + +export interface WeeklyFeatureStatus { + generation_enabled: boolean; + send_enabled: boolean; + replaces_monday_daily_send: boolean; + next_generate_at: string; + next_send_at: string; + generation_job_registered: boolean; + send_job_registered: boolean; + status_counts: Record; +} + +export interface RepairIncident { + incident_id: string; + fingerprint: string; + scope: string; + error_type: string; + status: string; + attempt_count: number; + cooldown_until?: string; + branch?: string; + commit_sha?: string; + pr_url?: string; + circuit_reason?: string; + updated_at?: string; +} + +export interface RepairSummary { + enabled: boolean; + queued: number; + active_fingerprint: string; + circuit_open: boolean; + circuit_until: string; + status_counts: Record; } export interface TemplateItem { @@ -587,9 +624,11 @@ export const getRecoveryBacklog = (lookbackDays = 30) => get(`/v2/recovery/backlog?lookback_days=${lookbackDays}`); export const confirmRecovery = (body: { expected_version: string; tasks: { run_date: string; group_id: number }[] }) => post<{ status: string; generation_only: boolean; send_invoked: boolean; results: { status: string; group_name?: string }[] }>("/v2/recovery/confirm", body); -export const listWeeklyInsights = () => get<{ schema_version: number; items: WeeklyInsight[] }>("/v2/weekly"); +export const listWeeklyInsights = () => get<{ schema_version: number; feature: WeeklyFeatureStatus; items: WeeklyInsight[] }>("/v2/weekly"); export const getWeeklyInsight = (weekStart: string, groupId: number) => get(`/v2/weekly/${weekStart}/${groupId}`); +export const listRepairIncidents = () => + get<{ schema_version: number; summary: RepairSummary; items: RepairIncident[] }>("/v2/repair/incidents"); export const retryFailed = (body: { group_id?: number; run_date?: string }) => post<{ results: { group_name?: string; status: string; detail?: string }[] }>("/v2/pipeline/retry-failed", body); export const pipelineGenerate = (body: { group_id?: number; run_date?: string; force?: boolean; refresh_messages?: boolean }) => diff --git a/frontend/src/pages/v2/Archive.tsx b/frontend/src/pages/v2/Archive.tsx index 0d6c028..01233bd 100644 --- a/frontend/src/pages/v2/Archive.tsx +++ b/frontend/src/pages/v2/Archive.tsx @@ -15,10 +15,14 @@ import { V2Run, V2RunDetail, WeeklyInsight, + WeeklyFeatureStatus, + RepairIncident, + RepairSummary, getArchiveGroups, getRunDetail, getWeeklyInsight, listWeeklyInsights, + listRepairIncidents, getV2File, readV2TextFile, restoreGroup, @@ -166,6 +170,17 @@ function preferredGroup(groups: ArchiveGroup[]): ArchiveGroup | undefined { })[0]; } +function repairStatusLabel(item: RepairIncident): string { + if (item.pr_url || item.status === "pr_created") return "修复 PR 待确认"; + if (item.status === "queued") return "等待 Codex 诊断"; + if (item.status === "running") return "Codex 正在维修"; + if (item.status === "environment") return "运行环境问题"; + if (item.status === "diagnostic_only") return item.scope === "send" || item.scope === "wechat" ? "发送人工复核" : "仅诊断，等待人工复核"; + if (item.status === "failed") return "自动维修失败"; + if (item.status === "corrupt") return "维修状态损坏"; + return item.status || "未知状态"; +} + export default function Archive() { const { msg, toast } = useToast(); const [groups, setGroups] = useState([]); @@ -186,6 +201,9 @@ export default function Archive() { const [weeklyItems, setWeeklyItems] = useState([]); const [selectedWeekly, setSelectedWeekly] = useState(null); const [weeklyLoading, setWeeklyLoading] = useState(true); + const [weeklyFeature, setWeeklyFeature] = useState(null); + const [repairItems, setRepairItems] = useState([]); + const [repairSummary, setRepairSummary] = useState(null); const loadArchive = async (): Promise => { setLoading(true); @@ -212,6 +230,7 @@ export default function Archive() { void loadArchive(); listWeeklyInsights() .then((response) => { + setWeeklyFeature(response.feature); const items = Array.isArray(response.items) ? response.items : []; setWeeklyItems(items); if (items[0]) return getWeeklyInsight(items[0].week_start, items[0].group_id); @@ -220,6 +239,12 @@ export default function Archive() { .then((detail) => detail && setSelectedWeekly(detail)) .catch((error) => toast(`周报归档读取失败：${safeError(error, "请稍后重试")}`)) .finally(() => setWeeklyLoading(false)); + listRepairIncidents() + .then((response) => { + setRepairItems(response.items || []); + setRepairSummary(response.summary); + }) + .catch((error) => toast(`自动维修状态读取失败：${safeError(error, "请稍后重试")}`)); }, []); const selectWeeklyInsight = (item: WeeklyInsight) => { @@ -402,7 +427,13 @@ export default function Archive() {

每周洞察

只读展示上一自然周的已保存日报聚合；不会重新上传整周聊天。

独立周报
- {weeklyLoading && weeklyItems.length === 0 ? : weeklyItems.length === 0 ? : ( + {weeklyFeature &&
+ 生成 {weeklyFeature.generation_enabled ? "已开启" : "未开启"} / 任务 {weeklyFeature.generation_job_registered ? "已注册" : "未注册"} + 发送 {weeklyFeature.send_enabled ? "已开启" : "未开启"} / 任务 {weeklyFeature.send_job_registered ? "已注册" : "未注册"} + 下次生成 {weeklyFeature.next_generate_at || "配置异常"} + 下次发送 {weeklyFeature.next_send_at || "配置异常"} +
} + {weeklyLoading && weeklyItems.length === 0 ? : weeklyItems.length === 0 ? : (
{weeklyItems.map((item) => (
+
+

自动维修

只处理脱敏故障事件；代码修复只会创建待确认 PR，不会自动合并、部署或补发。

{repairSummary?.circuit_open ? "已熔断" : repairSummary?.enabled ? "运行中" : "未开启"}
+
+ 等待处理 {repairSummary?.queued ?? 0} + 活跃指纹 {repairSummary?.active_fingerprint || "无"} + {repairSummary?.circuit_open && 熔断至 {repairSummary.circuit_until}} +
+ {repairItems.length === 0 ? : ( +
{repairItems.slice(0, 10).map((item) => ( + + ))}
+ )} +
+

群聊归档

选择群聊后，可通过月历查看有归档的日期。

diff --git a/frontend/src/pages/v2/Settings.tsx b/frontend/src/pages/v2/Settings.tsx index 5a96c40..580ad4c 100644 --- a/frontend/src/pages/v2/Settings.tsx +++ b/frontend/src/pages/v2/Settings.tsx @@ -113,6 +113,8 @@ const SETTING_GROUPS = [ ] as const; const CHECK_LABEL: Record = { + weekly_insights: "周报调度", + auto_repair: "Codex 自动维修", wechat_data_analysis: "WeChatDataAnalysis 数据源", codex_summary: "Codex GPT 群聊总结", deepseek_fallback: "DeepSeek 总结备用", diff --git a/scripts/codex_image_automation.py b/scripts/codex_image_automation.py index c3eeef9..1d63c38 100644 --- a/scripts/codex_image_automation.py +++ b/scripts/codex_image_automation.py @@ -334,51 +334,19 @@ def _sha256(path: Path) -> str: def _sync_scheduler_result(store: RunStore, group_name: str, run_date: str) -> None: - """同步人工认领结果，不触碰已经完成的邮件批次或微信发送字段。""" - scheduler_path = store.root / ".scheduler" / f"{run_date}.json" - if not scheduler_path.is_file(): - return + """从全部权威 run.json 重算批次状态，不增添幽灵群或覆盖发送字段。""" + del group_name + from app.scheduler.daily_v2_job import reconcile_daily_schedule_from_runs + + from app.config.settings import Settings + from app.scheduler.daily_v2_job import ScheduleStateCorruptionError + + settings = Settings(_env_file=None, output_root_override=str(store.root)) try: - state = json.loads(scheduler_path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): + reconcile_daily_schedule_from_runs(settings, run_date) + except ScheduleStateCorruptionError: + # 工件认领已经成功；损坏 scheduler 保持原样并继续在接口中显式暴露。 return - if not isinstance(state, dict): - return - raw_results = state.get("generation_results") - results = list(raw_results) if isinstance(raw_results, list) else [] - replacement = { - "group_name": group_name, - "status": "ready_to_send", - "detail": "图片已安全恢复，可以按原计划发送", - } - matched = False - for index, item in enumerate(results): - if isinstance(item, dict) and str(item.get("group_name") or "") == group_name: - results[index] = replacement - matched = True - break - if not matched: - results.append(replacement) - statuses = {str(item.get("status") or "") for item in results if isinstance(item, dict)} - if statuses and statuses <= {"ready_to_send", "skipped", "no_groups"}: - generation_status = "success" - elif statuses == {"failed"}: - generation_status = "failed" - else: - generation_status = "partial" - state.update( - generation_results=results, - generation_status=generation_status, - generation_recovered_at=datetime.now().astimezone().isoformat(), - updated_at=datetime.now().astimezone().isoformat(), - ) - scheduler_path.parent.mkdir(parents=True, exist_ok=True) - temp_path = scheduler_path.with_suffix(f".json.{os.getpid()}.{uuid.uuid4().hex}.tmp") - try: - temp_path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") - os.replace(temp_path, scheduler_path) - finally: - temp_path.unlink(missing_ok=True) def _new_images(marker: dict[str, Any]) -> list[Path]: diff --git a/scripts/repair_worker.py b/scripts/repair_worker.py new file mode 100644 index 0000000..2187edd --- /dev/null +++ b/scripts/repair_worker.py @@ -0,0 +1,26 @@ +"""独立运行一次 GroupBrief Codex 维修队列。""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) + +from app.config.settings import get_settings +from app.repair.controller import RepairController +from app.repair.events import capture_persisted_incidents + + +def main() -> int: + settings = get_settings() + capture_persisted_incidents(settings) + result = RepairController(settings).run_once() + print(json.dumps(result, ensure_ascii=False)) + return 0 if result.get("status") in {"disabled", "not_run", "pr_created"} else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_codex_image_automation.py b/tests/test_codex_image_automation.py index 40b9d07..d463fe1 100644 --- a/tests/test_codex_image_automation.py +++ b/tests/test_codex_image_automation.py @@ -256,13 +256,17 @@ def test_explicit_adopt_preserves_timing_audits_hash_and_syncs_scheduler(tmp_pat json.dumps( { "run_date": "2026-08-20", + "manifest_version": 1, + "manifest_created_at": "2026-08-20T00:14:00+08:00", + "expected_groups": [{"group_id": 1, "group_name": "测试群"}], "generation_status": "partial", - "generation_completed_at": "done", + "generation_completed_at": "2026-08-20T00:20:00+08:00", "generation_results": [ {"group_name": "测试群", "status": "failed", "error_type": IMAGE_GENERATION_FAILED}, {"group_name": "其他群", "status": "ready_to_send"}, ], - "email_completed_at": "sent-before-recovery", + "email_completed_at": "2026-08-20T00:30:00+08:00", + "email_status": "sent", }, ensure_ascii=False, ), @@ -281,5 +285,5 @@ def test_explicit_adopt_preserves_timing_audits_hash_and_syncs_scheduler(tmp_pat assert run["image_recovery"]["preserved_imagegen_ms"] is True assert len(run["image_recovery"]["sha256"]) == 64 assert scheduler["generation_status"] == "success" - assert scheduler["email_completed_at"] == "sent-before-recovery" + assert scheduler["email_completed_at"] == "2026-08-20T00:30:00+08:00" assert scheduler["generation_results"][0]["status"] == "ready_to_send" diff --git a/tests/test_reliability_watchdog.py b/tests/test_reliability_watchdog.py index 5bda9ea..3ce6f60 100644 --- a/tests/test_reliability_watchdog.py +++ b/tests/test_reliability_watchdog.py @@ -97,3 +97,49 @@ def send_due_for_dates(self, run_dates, *, now, recovery): ) assert calls == [] + + +def test_watchdog_does_not_scan_monday_daily_send_when_weekly_replaces_it( + tmp_path, + monkeypatch, +): + from app.config.settings import Settings + from app.scheduler import reliability_watchdog as watchdog + + settings = Settings( + _env_file=None, + reliability_watchdog_enabled=True, + reliability_lookback_days=1, + weekly_insights_enabled=True, + weekly_send_enabled=True, + weekly_replaces_monday_daily_send=True, + ) + + class TempState(watchdog.DailyScheduleState): + def __init__(self, _output_root): + super().__init__(tmp_path) + + class ForbiddenPipeline: + def __init__(self, settings): + raise AssertionError("周一启动恢复不得进入日报发送流水线") + + monkeypatch.setattr(watchdog, "DailyScheduleState", TempState) + monkeypatch.setattr(watchdog, "DailyPipeline", ForbiddenPipeline) + monkeypatch.setattr( + watchdog, + "run_daily_v2_job", + lambda *args, **kwargs: {"status": "success"}, + ) + + result = watchdog.run_reliability_watchdog( + settings=settings, + now=datetime(2026, 8, 31, 9, 0, tzinfo=ZoneInfo("Asia/Shanghai")), + ) + + assert result["status"] == "success" + assert result["send"] == [ + { + "status": "not_run", + "detail": "周一日报微信发送已由上一自然周周报替代", + } + ] diff --git a/tests/test_repair_api.py b/tests/test_repair_api.py new file mode 100644 index 0000000..dd79d4d --- /dev/null +++ b/tests/test_repair_api.py @@ -0,0 +1,37 @@ +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.api.v2_repair import router +from app.config.settings import Settings, get_settings +from app.repair.store import RepairIncidentStore + + +def test_repair_api_is_read_only_and_does_not_expose_error_body(tmp_path): + settings = Settings( + _env_file=None, + output_root_override=str(tmp_path / "output"), + repair_enabled=True, + ) + incident = RepairIncidentStore(settings).record( + scope="ranking", + error_type="RANKING_FAILED", + stage="render", + source_path="daily/2026-09-05/group-opaque/run.json", + error_summary="authorization=secret-value", + ) + app = FastAPI() + app.include_router(router, prefix="/api/v2") + app.dependency_overrides[get_settings] = lambda: settings + client = TestClient(app) + + listing = client.get("/api/v2/repair/incidents") + detail = client.get(f"/api/v2/repair/incidents/{incident['incident_id']}") + missing = client.get("/api/v2/repair/incidents/00000000000000000000000000000000") + + assert listing.status_code == 200 + assert listing.json()["summary"]["queued"] == 1 + assert detail.status_code == 200 + assert detail.json()["fingerprint"] == incident["fingerprint"] + assert "redacted_error_summary" not in detail.json() + assert "secret-value" not in detail.text + assert missing.status_code == 404 diff --git a/tests/test_repair_controller.py b/tests/test_repair_controller.py new file mode 100644 index 0000000..913e49b --- /dev/null +++ b/tests/test_repair_controller.py @@ -0,0 +1,116 @@ +import json +import subprocess +from pathlib import Path + +from app.config.settings import Settings +from app.repair.controller import CommandResult, RepairController +from app.repair.store import RepairIncidentStore + + +class FakeRunner: + def __init__(self): + self.calls = [] + + def run(self, argv, *, cwd, stdin="", timeout=300): + self.calls.append((list(argv), Path(cwd), stdin, timeout)) + if argv[:3] == ["git", "symbolic-ref", "refs/remotes/origin/HEAD"]: + return CommandResult(0, "refs/remotes/origin/master\n") + if argv[:3] == ["git", "worktree", "add"]: + Path(argv[5]).mkdir(parents=True) + (Path(argv[5]) / "app" / "repair").mkdir(parents=True) + (Path(argv[5]) / "app" / "repair" / "codex_result.schema.json").write_text("{}") + return CommandResult(0) + if len(argv) > 1 and argv[1] == "exec": + result_path = Path(argv[argv.index("--output-last-message") + 1]) + result_path.write_text(json.dumps({ + "regression_test_added": True, + "regression_test_reproduced": True, + "tests_passed": True, + "safe_to_propose_pr": True, + }), encoding="utf-8") + return CommandResult(0, '{"type":"thread.started","thread_id":"thread-test"}\n') + if argv[:3] == ["git", "status", "--porcelain"]: + return CommandResult(0, " M app/example.py\n?? tests/test_example.py\n") + if argv[:3] == ["git", "rev-parse", "HEAD"]: + return CommandResult(0, "abc123\n") + if argv[:3] == ["gh", "pr", "create"]: + return CommandResult(0, "https://github.test/pr/1\n") + return CommandResult(0, "2 passed\n") + + +class TimeoutRunner(FakeRunner): + def run(self, argv, *, cwd, stdin="", timeout=300): + if len(argv) > 1 and argv[1] == "exec": + self.calls.append((list(argv), Path(cwd), stdin, timeout)) + raise subprocess.TimeoutExpired(argv, timeout) + return super().run(argv, cwd=cwd, stdin=stdin, timeout=timeout) + + +def test_controller_uses_fixed_restricted_codex_command_and_only_creates_pr(tmp_path): + settings = Settings( + _env_file=None, + output_root_override=str(tmp_path / "output"), + repair_worktree_root=str(tmp_path / "worktrees"), + repair_enabled=True, + ) + store = RepairIncidentStore(settings) + store.record( + scope="ranking", + error_type="RANKING_FAILED", + stage="render", + source_path="run.json", + error_summary="authorization=top-secret", + ) + runner = FakeRunner() + repository = tmp_path / "repo" + repository.mkdir() + + result = RepairController( + settings, + store=store, + runner=runner, + repository_root=repository, + ).run_once() + + assert result["status"] == "pr_created" + codex_argv, _, prompt, timeout = next(call for call in runner.calls if len(call[0]) > 1 and call[0][1] == "exec") + assert ["--model", "gpt-5.6-sol"] == codex_argv[codex_argv.index("--model"):codex_argv.index("--model") + 2] + for flag in ("--sandbox", "--approve-for-me", "--json", "--output-schema", "--ephemeral", "--ignore-user-config"): + assert flag in codex_argv + assert "top-secret" not in prompt + assert timeout == 3600 + flattened = [call[0] for call in runner.calls] + assert not any(argv[:3] == ["gh", "pr", "merge"] for argv in flattened) + assert not any("deploy" in argv or "restart" in argv for argv in flattened) + + +def test_controller_timeout_is_persisted_and_never_pushes(tmp_path): + settings = Settings( + _env_file=None, + output_root_override=str(tmp_path / "output"), + repair_worktree_root=str(tmp_path / "worktrees"), + repair_enabled=True, + repair_timeout_minutes=1, + ) + store = RepairIncidentStore(settings) + store.record( + scope="ranking", + error_type="RANKING_TIMEOUT_TEST", + stage="render", + source_path="run.json", + ) + runner = TimeoutRunner() + repository = tmp_path / "repo" + repository.mkdir() + + result = RepairController( + settings, + store=store, + runner=runner, + repository_root=repository, + ).run_once() + + assert result["status"] == "failed" + assert "1 分钟" in result["last_error"] + assert not any(call[0][:2] == ["git", "push"] for call in runner.calls) + assert not any(call[0][:3] == ["gh", "pr", "create"] for call in runner.calls) diff --git a/tests/test_repair_store.py b/tests/test_repair_store.py new file mode 100644 index 0000000..64e4137 --- /dev/null +++ b/tests/test_repair_store.py @@ -0,0 +1,109 @@ +from datetime import datetime, timedelta +from zoneinfo import ZoneInfo + +from app.config.settings import Settings +from app.repair.store import RepairIncidentStore, repair_mode_for + + +def _settings(tmp_path, **updates): + return Settings( + _env_file=None, + output_root_override=str(tmp_path / "output"), + repair_enabled=True, + **updates, + ) + + +def test_incident_is_redacted_and_deduplicated_for_seven_days(tmp_path): + store = RepairIncidentStore(_settings(tmp_path)) + now = datetime(2026, 9, 5, 9, 0, tzinfo=ZoneInfo("Asia/Shanghai")) + first = store.record( + scope="ranking", + error_type="RANKING_FAILED", + stage="render", + source_path="output/group/run.json", + error_summary="token=secret-value render failed", + now=now, + ) + second = store.record( + scope="ranking", + error_type="RANKING_FAILED", + stage="render", + source_path="another/location.json", + error_summary="token=secret-value render failed", + now=now + timedelta(days=1), + ) + + assert first["incident_id"] == second["incident_id"] + assert second["occurrence_count"] == 2 + assert "secret-value" not in second["redacted_error_summary"] + + +def test_unknown_external_and_recursive_incidents_never_enter_code_fix_queue(tmp_path): + store = RepairIncidentStore(_settings(tmp_path)) + unknown = store.record( + scope="send", + error_type="SEND_RESULT_UNKNOWN", + stage="submit", + source_path="run.json", + ) + recursive = store.record( + scope="repair", + error_type="CODEX_EXEC_FAILED", + stage="controller", + source_path="controller.json", + ) + + assert unknown["status"] == "diagnostic_only" + assert recursive["status"] == "diagnostic_only" + assert repair_mode_for("wechat", "SEND_TEXT_FAILED") == "environment" + + +def test_daily_limit_and_three_failures_open_circuit_for_24_hours(tmp_path): + settings = _settings(tmp_path, repair_max_per_day=3) + store = RepairIncidentStore(settings) + now = datetime(2026, 9, 5, 9, 0, tzinfo=ZoneInfo("Asia/Shanghai")) + for index in range(3): + store.record( + scope="ranking", + error_type=f"RANKING_FAILED_{index}", + stage="render", + source_path=f"run-{index}.json", + now=now, + ) + incident, reason = store.start_next(now=now + timedelta(minutes=index)) + assert reason == "started" + store.finish(incident, success=False, reason="test failure", now=now + timedelta(minutes=index)) + + incident, reason = store.start_next(now=now + timedelta(minutes=4)) + assert incident is None + assert reason == "circuit_open" + summary = store.summary() + assert summary["circuit_open"] is True + assert datetime.fromisoformat(summary["circuit_until"]) >= now + timedelta(hours=23) + + +def test_daily_limit_stops_third_code_repair_attempt(tmp_path): + settings = _settings( + tmp_path, + repair_max_per_day=2, + repair_circuit_failure_threshold=99, + ) + store = RepairIncidentStore(settings) + now = datetime(2026, 9, 5, 10, 0, tzinfo=ZoneInfo("Asia/Shanghai")) + for index in range(3): + store.record( + scope="ranking", + error_type=f"DAILY_LIMIT_{index}", + stage="render", + source_path=f"run-{index}.json", + now=now, + ) + for index in range(2): + incident, reason = store.start_next(now=now + timedelta(minutes=index)) + assert reason == "started" + store.finish(incident, success=False, now=now + timedelta(minutes=index)) + + incident, reason = store.start_next(now=now + timedelta(minutes=3)) + assert incident is None + assert reason == "daily_limit" diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 3c1e498..c230a2a 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -9,6 +9,7 @@ from app.scheduler.manager import ( _parse_generate_time, _schedule_startup_recovery, + _schedule_weekly_replacement_jobs, get_scheduler, start_scheduler, stop_scheduler, @@ -55,6 +56,122 @@ def test_scheduler_jobs_configured(): assert get_scheduler() is None +def test_reconcile_generation_summary_uses_all_authoritative_runs(tmp_path): + from app.config.settings import Settings + from app.scheduler.daily_v2_job import ( + DailyScheduleState, + reconcile_daily_schedule_from_runs, + ) + from app.v2.constants import READY_TO_SEND + from app.v2.run_store import RunStore + + settings = Settings(_env_file=None, output_root_override=str(tmp_path / "output")) + schedule = DailyScheduleState(settings.output_dir) + schedule.update( + "2026-09-05", + manifest_version=1, + manifest_created_at="2026-09-05T00:15:00+08:00", + expected_groups=[ + {"group_id": 1, "group_name": "群一"}, + {"group_id": 2, "group_name": "群二"}, + ], + generation_started_at="2026-09-05T00:15:00+08:00", + generation_completed_at="2026-09-05T00:20:00+08:00", + generation_status="partial", + generation_results=[{"group_name": "群一", "status": "failed"}], + email_recovery_required=False, + ) + runs = RunStore(settings.output_dir) + for group_id, group_name in ((1, "群一"), (2, "群二")): + runs.save_run( + group_name, + "2026-09-05", + { + "group_id": str(group_id), + "status": READY_TO_SEND, + "image_recovered_at": "2026-09-05T09:00:00+08:00", + }, + ) + + state = reconcile_daily_schedule_from_runs(settings, "2026-09-05") + + assert state["generation_status"] == "success" + assert [item["group_name"] for item in state["generation_results"]] == ["群一", "群二"] + assert all(item["status"] == "ready_to_send" for item in state["generation_results"]) + assert state["email_recovery_required"] is False + + +def test_scheduler_uses_daily_batch_as_the_only_monday_send_entry(): + from app.config.settings import Settings + + settings = Settings( + _env_file=None, + reliability_watchdog_enabled=False, + weekly_insights_enabled=True, + weekly_send_enabled=True, + weekly_replaces_monday_daily_send=True, + ) + try: + start_scheduler(settings) + scheduler = get_scheduler() + assert scheduler is not None + ids = {job.id for job in scheduler.get_jobs()} + assert "daily_wechat_send_batch" in ids + assert "weekly_insights_generate" in ids + assert "weekly_insights_send" not in ids + finally: + stop_scheduler() + + +def test_scheduler_registers_independent_repair_controller_only_when_enabled(): + from app.config.settings import Settings + + settings = Settings( + _env_file=None, + reliability_watchdog_enabled=False, + repair_enabled=True, + repair_poll_interval_minutes=7, + ) + try: + start_scheduler(settings) + scheduler = get_scheduler() + job = scheduler.get_job("repair_controller") + assert job is not None + assert job.name == "RepairController" + assert int(job.trigger.interval.total_seconds()) == 7 * 60 + finally: + stop_scheduler() + + +def test_monday_daily_manifest_finishes_at_ready_when_weekly_replaces_send(): + from app.db.models import Group + from app.scheduler.task_manifest import build_expected_groups + + group = Group( + id=7, + display_name="测试群", + wechat_group_id="test@chatroom", + wechat_send_enabled=True, + ) + monday = build_expected_groups( + [group], + datetime(2026, 8, 31).date(), + timezone="Asia/Shanghai", + weekly_replaces_monday_daily_send=True, + )[0] + tuesday = build_expected_groups( + [group], + datetime(2026, 9, 1).date(), + timezone="Asia/Shanghai", + weekly_replaces_monday_daily_send=True, + )[0] + + assert monday["expected_terminal"] == "READY_TO_SEND" + assert monday["wechat_send_replaced_by_weekly"] is True + assert tuesday["expected_terminal"] == "SENT" + assert tuesday["wechat_send_replaced_by_weekly"] is False + + def test_scheduler_registers_only_one_startup_recovery_job(): from app.config.settings import Settings @@ -869,6 +986,125 @@ def add_job(self, func, **kwargs): assert held_scheduler.jobs == {} +def test_monday_replacement_never_reschedules_daily_wechat_send(tmp_path): + from app.config.settings import Settings + from app.scheduler.manager import _schedule_on_demand_jobs + from app.v2.constants import READY_TO_SEND + from app.v2.run_store import RunStore + + settings = Settings( + _env_file=None, + output_root_override=str(tmp_path / "output"), + weekly_insights_enabled=True, + weekly_send_enabled=True, + weekly_replaces_monday_daily_send=True, + ) + store = RunStore(settings.output_dir) + run_date = "2026-08-31" + store.save_run( + "周一日报群", + run_date, + { + "group_id": "1", + "status": READY_TO_SEND, + "wechat_send_enabled": True, + "send_hold": False, + }, + ) + + class FakeScheduler: + def __init__(self): + self.jobs = {} + + def add_job(self, func, **kwargs): + self.jobs[kwargs["id"]] = (func, kwargs) + + scheduler = FakeScheduler() + scheduled = _schedule_on_demand_jobs( + scheduler, + settings, + now=datetime(2026, 8, 31, 8, 40, tzinfo=ZoneInfo("Asia/Shanghai")), + run_dates=[run_date], + ) + + assert scheduled == [] + assert scheduler.jobs == {} + + +def test_send_batch_routes_monday_to_weekly_and_other_days_to_daily(monkeypatch): + from app.config.settings import Settings + from app.scheduler import manager + + settings = Settings( + _env_file=None, + weekly_insights_enabled=True, + weekly_send_enabled=True, + weekly_replaces_monday_daily_send=True, + ) + calls = [] + monday = datetime(2026, 8, 31, 8, 30, tzinfo=ZoneInfo("Asia/Shanghai")) + monkeypatch.setattr(manager, "get_settings", lambda: settings) + monkeypatch.setattr(manager, "_normalize_now", lambda _settings, now=None: now or monday) + monkeypatch.setattr( + manager, + "run_scheduled_weekly_send", + lambda **kwargs: calls.append(("weekly", kwargs["now"])) or {"status": "success"}, + ) + monkeypatch.setattr( + manager, + "run_send_due_job", + lambda **kwargs: calls.append(("daily", kwargs["run_date"])) or {"status": "success"}, + ) + + manager.run_scheduled_send_batch("2026-08-31") + assert calls == [("weekly", monday)] + + tuesday = datetime(2026, 9, 1, 8, 30, tzinfo=ZoneInfo("Asia/Shanghai")) + monkeypatch.setattr(manager, "_normalize_now", lambda _settings, now=None: now or tuesday) + manager.run_scheduled_send_batch("2026-09-01") + assert calls[-1] == ("daily", "2026-09-01") + + +def test_monday_restart_rebuilds_weekly_generation_and_send(monkeypatch): + from app.config.settings import Settings + from app.scheduler import manager + + settings = Settings( + _env_file=None, + weekly_insights_enabled=True, + weekly_send_enabled=True, + weekly_replaces_monday_daily_send=True, + ) + + class FakeStore: + def list_states(self): + return [ + { + "week_start": "2026-08-24", + "week_end": "2026-08-30", + "status": "ready_to_send", + } + ] + + class FakeScheduler: + def __init__(self): + self.jobs = {} + + def add_job(self, func, **kwargs): + self.jobs[kwargs["id"]] = (func, kwargs) + + monkeypatch.setattr(manager, "WeeklyStore", lambda _output_dir: FakeStore()) + scheduler = FakeScheduler() + scheduled = _schedule_weekly_replacement_jobs( + scheduler, + settings, + now=datetime(2026, 8, 31, 8, 40, tzinfo=ZoneInfo("Asia/Shanghai")), + ) + + assert scheduled == ["weekly_generate_once_20260831", "weekly_send_once_20260831"] + assert set(scheduler.jobs) == set(scheduled) + + def test_on_demand_rebuilds_only_persisted_generation_and_send_retries(tmp_path): from app.config.settings import Settings from app.scheduler.daily_v2_job import DailyScheduleState diff --git a/tests/test_v2_image_task.py b/tests/test_v2_image_task.py index 7c4e63e..f614bac 100644 --- a/tests/test_v2_image_task.py +++ b/tests/test_v2_image_task.py @@ -297,7 +297,6 @@ def generate(self, prompt_file, output_path, **_kwargs): job = _job(tmp_path, "纠错重画群", generator) verification_results = iter( [ - (False, "图片文件不存在"), (False, "图片事实校验失败：无证据数字：45元, 11218"), (True, "OK"), ] @@ -438,6 +437,19 @@ def test_skip_when_image_exists(tmp_path): job = _job(tmp_path, "群1", gen) # 先成功生成一次 assert job.run()["status"] == "success" + (job.output_path.parent / "run.json").write_text( + json.dumps( + { + "image_enabled": True, + "image_fallback_level": 0, + "image_variant": "normal", + "image_status": "success", + "image_job": {"status": "completed"}, + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) # 再跑：已存在有效图片 → 跳过（不重复生成） result = job.run() assert result["status"] == "skipped" diff --git a/tests/test_v2_pipeline.py b/tests/test_v2_pipeline.py index 0b6542d..b8dfc63 100644 --- a/tests/test_v2_pipeline.py +++ b/tests/test_v2_pipeline.py @@ -32,6 +32,7 @@ IMAGE_FALLBACK_NOT_SENDABLE, IMAGE_FILE_MISSING, IMAGE_GENERATION_FAILED, + IMAGE_PROVENANCE_MISSING, IMAGE_READY, PROMPT_READY, READY_TO_SEND, @@ -1110,6 +1111,9 @@ def _ready_run_contract(*, image_enabled: bool = False) -> dict: }, "prompt_stale": False, "image_stale": False, + "image_fallback_level": 0, + "image_variant": "normal", + "image_status": "success", } @@ -1138,6 +1142,29 @@ def test_send_due_sends_text_then_image(tmp_path): assert len(sender.image_calls) == 1 +def test_monday_weekly_replacement_blocks_direct_daily_send_due(tmp_path): + pipeline = _ready_to_send(tmp_path) + pipeline.settings = pipeline.settings.model_copy( + update={ + "weekly_insights_enabled": True, + "weekly_send_enabled": True, + "weekly_replaces_monday_daily_send": True, + } + ) + monday = "2026-08-31" + original = pipeline.store.load_run("测试群", "2026-08-18") + pipeline.store.save_run("测试群", monday, original) + + results = pipeline.send_due( + now=datetime(2026, 8, 31, 8, 30, tzinfo=ZoneInfo("Asia/Shanghai")) + ) + + assert results == [] + assert pipeline.sender.text_calls == [] + assert pipeline.sender.image_calls == [] + assert pipeline.store.load_run("测试群", monday)["status"] == READY_TO_SEND + + @pytest.mark.parametrize( ("fallback_level", "image_variant"), [(3, "normal"), (0, "pillow")], @@ -1309,6 +1336,41 @@ def test_explicit_text_failures_use_backoff_and_stop_after_send_retry_budget(tmp assert len(sender.text_calls) == 3 +@pytest.mark.parametrize( + ("detail", "error_type"), + [ + ("群聊分区未找到目标（匹配数 0；可信分区 0）", "WECHAT_TARGET_NOT_FOUND"), + ("UIA 精确群聊项数量不是 1（当前 2）", "WECHAT_TARGET_AMBIGUOUS"), + ("文字粘贴后未观察到输入区暂存，已停止且未按 Enter", "WECHAT_TEXT_NOT_STAGED"), + ], +) +def test_deterministic_wechat_pre_submit_failure_holds_on_first_attempt( + tmp_path, detail, error_type +): + class PreSubmitFailureSender(FakeSender): + def send_text(self, target, text): + from app.sender.base import SendResult + + self.text_calls.append((target, text)) + return SendResult(False, detail, datetime.now().isoformat(), submitted=False) + + sender = PreSubmitFailureSender() + pipeline, _ = _make_pipeline(tmp_path, sender=sender, image_enabled=False) + pipeline.generate_all(run_date="2026-08-18") + + first = pipeline.send_due(now=datetime(2026, 8, 18, 8, 31, 0)) + second = pipeline.send_due(now=datetime(2026, 8, 18, 8, 32, 0)) + run = pipeline.store.load_run("测试群", "2026-08-18") + + assert first[0]["status"] == "held" + assert first[0]["error_type"] == error_type + assert second == [] + assert len(sender.text_calls) == 1 + assert run["send_state"] == "failed_final" + assert run["send_hold"] is True + assert run["send_retry_attempt_count"] == 0 + + def test_explicit_unsubmitted_final_failure_can_be_reset_without_sending(tmp_path): pipeline = _ready_to_send(tmp_path, image_enabled=False) sender = pipeline.sender @@ -1916,6 +1978,35 @@ def test_submitted_but_unverified_text_is_held_without_retry(tmp_path): assert run["send_hold"] is True +def test_image_preview_not_observed_holds_without_retry_or_second_submit(tmp_path): + class PreviewFailureSender(FakeSender): + def send_image(self, target, image_path): + from app.sender.base import SendResult + + self.image_calls.append((target, Path(image_path))) + return SendResult( + False, + "图片粘贴后未观察到预览，已停止且未按 Enter", + datetime.now().isoformat(), + submitted=False, + ) + + pipeline = _ready_to_send(tmp_path) + sender = PreviewFailureSender() + pipeline.sender = sender + + first = pipeline.send_due(now=datetime(2026, 8, 18, 8, 31, 0)) + second = pipeline.send_due(now=datetime(2026, 8, 18, 8, 32, 0)) + run = pipeline.store.load_run("测试群", "2026-08-18") + + assert first[0]["error_type"] == "WECHAT_IMAGE_NOT_STAGED" + assert second == [] + assert len(sender.text_calls) == 1 + assert len(sender.image_calls) == 1 + assert run["send_state"] == "failed_final" + assert run["send_hold"] is True + + def test_submitted_failure_is_still_held_as_unknown(tmp_path): pipeline = _ready_to_send(tmp_path) sender = SubmittedFailureTextSender() @@ -2181,4 +2272,4 @@ def test_run_store_send_claim_treats_legacy_artifact_without_contract_as_stale(t ) assert claim_id is None - assert reason == "artifact_stale" + assert reason == IMAGE_PROVENANCE_MISSING diff --git a/tests/test_v2_ui_router_contract.py b/tests/test_v2_ui_router_contract.py index 477a48d..6098ed1 100644 --- a/tests/test_v2_ui_router_contract.py +++ b/tests/test_v2_ui_router_contract.py @@ -45,6 +45,8 @@ ("GET", "/api/v2/weekly", "list_weekly_insights_api_v2_weekly_get"), ("GET", "/api/v2/weekly/{week_start}/{group_id}", "weekly_insight_detail_api_v2_weekly__week_start___group_id__get"), ("GET", "/api/v2/weekly/{week_start}/{group_id}/card", "weekly_insight_card_api_v2_weekly__week_start___group_id__card_get"), + ("GET", "/api/v2/repair/incidents", "list_repair_incidents_api_v2_repair_incidents_get"), + ("GET", "/api/v2/repair/incidents/{incident_id}", "repair_incident_detail_api_v2_repair_incidents__incident_id__get"), } diff --git a/tests/test_weekly_insights.py b/tests/test_weekly_insights.py index 0099233..1140812 100644 --- a/tests/test_weekly_insights.py +++ b/tests/test_weekly_insights.py @@ -162,6 +162,7 @@ def test_weekly_ai_failure_still_creates_deterministic_text_and_card(tmp_path): state = weekly_store.load("2026-08-24", "2026-08-30", group.id) assert state["status"] == "ready_to_send" assert state["ai_status"] == "failed" + assert state["error_type"] == "WEEKLY_AI_FAILED_FALLBACK" assert state["narrative_source"] == "local_deterministic" assert "周度洞察" in state["narrative"] @@ -232,3 +233,84 @@ def test_stale_weekly_send_claim_becomes_manual_hold_without_resubmit(tmp_path): assert sender.calls == [] assert state["status"] == "needs_attention" assert state["send_claim_id"] == "" + + +def test_weekly_corrupt_state_is_visible_in_archive(tmp_path): + store = WeeklyStore(tmp_path / "output") + path = store.state_path("2026-08-24", "2026-08-30", 7) + path.parent.mkdir(parents=True) + path.write_text("{broken", encoding="utf-8") + + states = store.list_states() + + assert len(states) == 1 + assert states[0]["status"] == "needs_attention" + assert states[0]["error_type"] == "WEEKLY_STATE_CORRUPT" + + +def test_weekly_hash_mismatch_holds_before_sender_call(tmp_path): + settings = _settings(tmp_path, weekly_send_enabled=True) + group = _group(settings, send=True) + daily_store = RunStore(tmp_path / "output") + weekly_store = WeeklyStore(daily_store.root) + _daily(daily_store, group, "2026-08-24", count=2, identity="a", name="成员A") + sender = FakeSender() + service = WeeklyInsightsService( + settings, + daily_store=daily_store, + weekly_store=weekly_store, + provider_factory=lambda _settings: FakeProvider([]), + sender=sender, + ) + service.generate_previous_week( + now=datetime(2026, 8, 31, 7, 45, tzinfo=ZoneInfo("Asia/Shanghai")) + ) + weekly_store.text_path("2026-08-24", "2026-08-30", group.id).write_text( + "被篡改的周报", encoding="utf-8" + ) + + result = service.send_due( + now=datetime(2026, 8, 31, 8, 30, tzinfo=ZoneInfo("Asia/Shanghai")) + ) + state = weekly_store.load("2026-08-24", "2026-08-30", group.id) + + assert sender.calls == [] + assert result[0]["error_type"] == "WEEKLY_ARTIFACT_HASH_MISMATCH" + assert state["status"] == "needs_attention" + + +def test_weekly_group_failure_does_not_block_another_group(tmp_path, monkeypatch): + settings = _settings(tmp_path) + first = _group(settings) + with Session(repo.engine) as session: + second = repo.save_group( + session, + Group( + display_name="第二周报群", + wechat_group_id="weekly-2@chatroom", + wechat_group_name="第二周报群", + summary_provider="codex", + summary_model="gpt-5.6-sol", + ), + ) + service = WeeklyInsightsService(settings, sender=FakeSender()) + original = service._generate_group + + def fail_first(group, *args): + if group.id == first.id: + raise ValueError("message_count 不是合法整数") + return { + "group_id": second.id, + "group_name": second.display_name, + "status": "ready_to_send", + } + + monkeypatch.setattr(service, "_generate_group", fail_first) + result = service.generate_previous_week( + now=datetime(2026, 8, 31, 7, 45, tzinfo=ZoneInfo("Asia/Shanghai")) + ) + + assert result["status"] == "partial" + assert [item["status"] for item in result["results"]] == ["held", "ready_to_send"] + held = service.store.load("2026-08-24", "2026-08-30", first.id) + assert held["error_type"] == "WEEKLY_DATA_INVALID"