feat: 启用周报闭环并新增 Codex 自动维修

.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，以及明确未执行的生产步骤。

.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

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)

app/api/v2_ui.py

@@ -42,20 +42,63 @@
 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
 
 
 router = APIRouter(prefix="/api/v2", tags=["v2-ui"])
 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)
 
 
 @router.get("/system/health")
 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)

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}")

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"

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

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)
 
 

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)