fix: 外部审查 P1/P2 批量修复（16 项）

core/brief.go

@@ -532,9 +532,8 @@ func HandleSendBriefEmailEffect(ctx context.Context, e *HandlerEnv, task *Task)
 		).Scan(&htmlContent)
 	}
 	if !SendEmailHTML(e.SMTP, recipient, subject, htmlContent) {
-		return map[string]any{
-			"sent": false, "error": "SMTP 未配置或发送失败（账本未登记，可重试）", "effect_key": effectKey,
-		}, nil
+		// R22：发送失败必须以 error 结束任务——nil error 会让任务假成功终结
+		return nil, fmt.Errorf("SMTP 未配置或发送失败（账本未登记，可重试）")
 	}
 	RegisterEffect(e.Store, effectKey, "mail_send", task.TaskID)
 	return map[string]any{"sent": true, "effect_key": effectKey}, nil

core/effects.go

@@ -113,14 +113,17 @@ func HasEffect(store *CoreStore, effectKey string) bool {
 }
 
 // RegisterEffect 登记副作用（唯一约束兜底并发；已存在返回 false）。
+// R13：task_effects.task_id 外键指向旧 tasks.id(32)——core_tasks 的 36 位 UUID
+// 写入必然 FK 失败且被忽略，导致邮件去重键永不登记。改为不关联旧执行体系，
+// 幂等键本身已足够去重。
 func RegisterEffect(store *CoreStore, effectKey, kind string, taskID string) bool {
 	if HasEffect(store, effectKey) {
 		return false
 	}
 	_, err := store.DB.Exec(
-		`INSERT INTO task_effects (id, task_id, effect_key, kind, payload, committed_at)
-		 VALUES ($1, $2, $3, $4, '{}', $5)`,
-		newCoreID(), nullIfEmpty(taskID), effectKey, kind, nowParam())
+		`INSERT INTO task_effects (id, effect_key, kind, payload, committed_at)
+		 VALUES ($1, $2, $3, '{}', $4)`,
+		newCoreID(), effectKey, kind, nowParam())
 	return err == nil
 }
 

core/handlers_paper.go

@@ -12,6 +12,7 @@ import (
 	"math"
 	"os"
 	"os/exec"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -123,8 +124,7 @@ func extractPDFTextRaw(pdfPath string, maxPages int) string {
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
 	defer cancel()
-	pages := fmt.Sprintf("1-%d", maxPages)
-	out, err := exec.CommandContext(ctx, pdftotext, "-f", "1", "-l", pages, "-layout", pdfPath, "-").Output()
+	out, err := exec.CommandContext(ctx, pdftotext, "-f", "1", "-l", strconv.Itoa(maxPages), "-layout", pdfPath, "-").Output()
 	if err != nil {
 		return ""
 	}
@@ -483,9 +483,23 @@ func HandleExtractClaims(ctx context.Context, env *HandlerEnv, task *Task) (map[
 			if !ok {
 				continue
 			}
-			if strings.TrimSpace(stringOf(m["statement"])) != "" {
-				items = append(items, m)
+			if strings.TrimSpace(stringOf(m["statement"])) == "" {
+				continue
+			}
+			// R12：对齐 apply 契约——quote 必填（原文逐字），kind 固定 text_passage，
+			// certainty 归一到真实枚举（tentative/supported）
+			norm := map[string]any{
+				"statement":     stringOf(m["statement"]),
+				"statement_zh":  stringOf(m["statement_zh"]),
+				"quote":         stringOf(m["quote"]),
+				"kind":          "text_passage",
 			}
+			if cert := stringOf(m["certainty"]); cert == "tentative" || cert == "supported" {
+				norm["certainty"] = cert
+			} else {
+				norm["certainty"] = "tentative"
+			}
+			items = append(items, norm)
 		}
 	}
 	trace := map[string]any{
@@ -510,13 +524,12 @@ func HandleExtractClaims(ctx context.Context, env *HandlerEnv, task *Task) (map[
 func buildClaimsPrompt(title, text string) string {
 	return "你是研究助理。请从以下论文文本中抽取最多 6 条可验证的核心判断（claims）。" +
 		"输出严格 JSON：\n" +
-		`{"claims":[{"statement":"英文判断句","certainty":"low|medium|high","evidence_quote":"原文引句","evidence_kind":"finding|method|dataset|benchmark"}]}` + "\n" +
-		"要求：statement 必须是论文明确支持的可检验判断；evidence_quote 必须是原文逐字摘录。\n" +
+		`{"claims":[{"statement":"英文判断句","statement_zh":"中文表述","certainty":"tentative|supported","quote":"原文逐字摘录","kind":"text_passage"}]}` + "\n" +
+		"要求：statement 必须是论文明确支持的可检验判断；quote 必须是文本中逐字出现的原句（用于定位验证）；" +
+		"certainty 只能是 tentative 或 supported。\n" +
 		fmt.Sprintf("论文标题: %s\n文本: %s\n", title, truncateStr(text, 6000))
 }
 
-// ---------- download_source / upsert_paper / fetch_topic_papers ----------
-
 func HandleDownloadSource(ctx context.Context, env *HandlerEnv, task *Task) (map[string]any, error) {
 	arxivID, _ := task.Input["arxiv_id"].(string)
 	if arxivID == "" {

core/runner.go

@@ -158,11 +158,9 @@ func (r *Runner) runTask(ctx context.Context, task *Task, pool *classPool) {
 	hbWG.Add(1)
 	go func() {
 		defer hbWG.Done()
-		// 续约节奏：lease 的 1/3（Python 同语义），至少 30s
-		interval := time.Duration(task.TimeoutS) * time.Second / 3
-		if interval < 30*time.Second {
-			interval = 30 * time.Second
-		}
+		// 续约节奏：HeartbeatTask 固定续约 +600s——周期必须明显小于续约窗口，
+		// 否则长任务第二次心跳前 lease 已被 reclaim（R27）。取 200s（600/3）。
+		interval := 200 * time.Second
 		ticker := time.NewTicker(interval)
 		defer ticker.Stop()
 		for {

core/scheduler.go

@@ -114,12 +114,17 @@ func (s *Scheduler) dailyBriefLoop(ctx context.Context) {
 		// cron 从配置读（默认 04:00 UTC = 北京 12:00）
 		hour := 4
 		var cronExpr string
-		var sendEmail bool
+		var sendEmail, enabled bool
 		if err := s.store.DB.QueryRow(
-			`SELECT COALESCE(cron_expression,''), COALESCE(send_email_report,false) FROM daily_report_configs LIMIT 1`,
-		).Scan(&cronExpr, &sendEmail); err == nil && cronExpr != "" {
+			`SELECT COALESCE(cron_expression,''), COALESCE(send_email_report,false), COALESCE(enabled,false)
+			 FROM daily_report_configs LIMIT 1`,
+		).Scan(&cronExpr, &sendEmail, &enabled); err == nil && cronExpr != "" {
 			hour = parseDailyCronHour(cronExpr)
 		}
+		// R14：日报开关关闭则完全不生成
+		if !enabled {
+			continue
+		}
 		if now.Hour() != hour {
 			continue
 		}

server/content.go

@@ -264,7 +264,7 @@ func (s *Server) ragAsk(ctx context.Context, question string, topK int) (string,
 	prompt := "请基于上下文回答问题，输出严格 JSON：" +
 		`{"answer":"...", "confidence":0.0}` + "\n" +
 		fmt.Sprintf("问题: %s\n上下文:\n%s", question, strings.Join(joined, "\n\n"))
-	parsed, res, err := s.gateway.CompleteJSON(ctx, "skim", prompt)
+	parsed, res, err := s.GW().CompleteJSON(ctx, "skim", prompt)
 	if err != nil {
 		return "", nil, nil, err
 	}
@@ -629,7 +629,7 @@ func (s *Server) handleWritingProcessMultimodal(w http.ResponseWriter, r *http.R
 
 // gatewayChat 网关单轮补全（writing 等简单场景）。
 func (s *Server) gatewayChat(ctx context.Context, prompt string) (string, error) {
-	res, err := s.gateway.Chat(ctx, "deep", prompt, 2048)
+	res, err := s.GW().Chat(ctx, "deep", prompt, 2048)
 	if err != nil {
 		return "", err
 	}

server/graph.go

@@ -835,16 +835,13 @@ func toAny(v any) any { return v }
 // loadVectors 加载论文 embedding（limit 篇，含 topic 标签）。
 func (s *Server) loadVectors(limit int, topicID string) ([]string, [][]float64, map[string][]string) {
 	query := `SELECT p.id, p.embedding_vec FROM papers p`
-	if topicID != "" {
-		query += ` JOIN paper_topics pt ON pt.paper_id = p.id AND pt.topic_id=$2`
-	}
-	query += fmt.Sprintf(` WHERE p.embedding_vec IS NOT NULL ORDER BY p.created_at DESC LIMIT $%d`, map[bool]int{true: 1, false: 2}[topicID != ""])
 	args := []any{}
 	if topicID != "" {
-		args = append(args, 50000, topicID)
-	} else {
-		args = append(args, limit)
+		query += ` JOIN paper_topics pt ON pt.paper_id = p.id AND pt.topic_id=$1`
+		args = append(args, topicID)
 	}
+	args = append(args, limit)
+	query += fmt.Sprintf(` WHERE p.embedding_vec IS NOT NULL ORDER BY p.created_at DESC LIMIT $%d`, len(args))
 	rows, err := s.db.Query(query, args...)
 	if err != nil {
 		return nil, nil, nil

server/handlers.go

@@ -3,6 +3,7 @@ package main
 // handlers.go：已移植路由的业务处理（folder-stats / 设备信息与授权决策）。
 
 import (
+	"encoding/json"
 	"net/http"
 	"time"
 )
@@ -80,7 +81,7 @@ func (s *Server) requireAuth(next http.HandlerFunc) http.HandlerFunc {
 			return
 		}
 		// API 令牌（SHA-256 查表）优先；Web JWT 兜底
-		if s.validAPIToken(token) {
+		if s.validAPITokenScoped(token, r.Method) {
 			next(w, r)
 			return
 		}
@@ -119,6 +120,34 @@ func (s *Server) validAPIToken(raw string) bool {
 	return err == nil
 }
 
+// validAPITokenScoped：API 令牌 + 方法级 scope 校验（R01——写方法需 write scope）。
+func (s *Server) validAPITokenScoped(raw string, method string) bool {
+	if method == http.MethodGet || method == http.MethodHead || method == http.MethodOptions {
+		return s.validAPIToken(raw)
+	}
+	var id string
+	var scopes []byte
+	err := s.db.QueryRow(
+		`SELECT id, scopes FROM api_tokens
+		 WHERE token_hash = $1 AND revoked_at IS NULL
+		   AND (expires_at IS NULL OR expires_at > now()::timestamp)`,
+		hashToken(raw),
+	).Scan(&id, &scopes)
+	if err != nil {
+		return false
+	}
+	var list []string
+	if json.Unmarshal(scopes, &list) != nil {
+		return false
+	}
+	for _, sc := range list {
+		if sc == "write" {
+			return true
+		}
+	}
+	return false
+}
+
 func validWebJWT(token, secret string) bool {
 	parsed, err := jwtParseHS256(token, secret)
 	return err == nil && parsed.valid

server/jobs.go

@@ -4,6 +4,7 @@ package main
 // Go Core 本身拥有 Job/Task/Attempt 权威数据——大部分端点直接查 PG core_jobs/core_tasks。
 
 import (
+	core "github.com/Color2333/PaperMind/core"
 	"encoding/json"
 	"net/http"
 )
@@ -70,6 +71,30 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	attempts := []map[string]any{}
+	attemptRows, err := s.db.Query(
+		`SELECT a.id, a.task_id, a.attempt_no, a.executor_id, a.status,
+			COALESCE(a.error_class,''), COALESCE(a.error_message,''), a.started_at::text,
+			COALESCE(TO_CHAR(a.finished_at,'YYYY-MM-DD"T"HH24:MI:SS"Z"'),'')
+		 FROM core_attempts a JOIN core_tasks t ON t.id = a.task_id
+		 WHERE t.job_id=$1 ORDER BY a.started_at DESC LIMIT 200`, jobID)
+	if err == nil {
+		for attemptRows.Next() {
+			var aid, tid, executor, aStatus, started, finished, errClass, errMsg string
+			var attemptNo int
+			if err := attemptRows.Scan(&aid, &tid, &attemptNo, &executor, &aStatus,
+				&errClass, &errMsg, &started, &finished); err == nil {
+				attempts = append(attempts, map[string]any{
+					"id": aid, "task_id": tid, "attempt_no": attemptNo,
+					"executor_id": executor, "status": aStatus,
+					"error_class": errClass, "error_message": errMsg,
+					"started_at": started, "finished_at": finished,
+				})
+			}
+		}
+		attemptRows.Close()
+	}
+
 	tasks := []map[string]any{}
 	taskRows, err := s.db.Query(
 		`SELECT id, capability, status, attempt_count, max_attempts,
@@ -96,7 +121,7 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {
 	json.Unmarshal([]byte(payload), &payloadObj)
 	writeJSON(w, http.StatusOK, map[string]any{
 		"id": jid, "kind": kind, "capability": capability, "status": jobStatus,
-		"created_at": createdAt, "payload": payloadObj, "tasks": tasks,
+		"created_at": createdAt, "payload": payloadObj, "tasks": tasks, "attempts": attempts,
 	})
 }
 
@@ -145,18 +170,15 @@ func (s *Server) handleSubmitDurable(w http.ResponseWriter, r *http.Request) {
 // handleCancelJob：POST /jobs/{id}/cancel。
 func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) {
 	jobID := r.PathValue("job_id")
-	res, err := s.db.Exec(
-		`UPDATE core_jobs SET status='cancelled', finished_at=now()::timestamp
-		 WHERE id=$1 AND status IN ('queued','running')`, jobID)
+	// R09：复用 CoreStore.CancelJob 权威状态机——leased 子任务进入 cancelling，
+	// Runner 心跳探测到 cancel_requested 后安全点退出，而不是只取消 queued。
+	store := &core.CoreStore{DB: s.db, IsPG: true}
+	counts, err := store.CancelJob(jobID)
 	if err != nil {
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"detail": err.Error()})
 		return
 	}
-	n, _ := res.RowsAffected()
-	// 取消未领取的子任务
-	_, _ = s.db.Exec(
-		`UPDATE core_tasks SET status='cancelled' WHERE job_id=$1 AND status='queued'`, jobID)
-	writeJSON(w, http.StatusOK, map[string]int{"cancelled": int(n)})
+	writeJSON(w, http.StatusOK, counts)
 }
 
 // handleRetryJob：POST /jobs/{id}/retry。
@@ -200,14 +222,14 @@ func (s *Server) handleRetryTask(w http.ResponseWriter, r *http.Request) {
 // handlePauseQueue / handleResumeQueue：队列暂停/恢复（system_flags 持久化）。
 func (s *Server) handlePauseQueue(w http.ResponseWriter, r *http.Request) {
 	_, _ = s.db.Exec(
-		`INSERT INTO system_flags (key, value, updated_at) VALUES ('queue_paused', 'true', now()::timestamp)
+		`INSERT INTO system_flags (key, value, updated_at) VALUES ('queue_paused', '1', now()::timestamp) ON CONFLICT (key) DO UPDATE SET value='1', updated_at=now()::timestamp
 		 ON CONFLICT (key) DO UPDATE SET value='true', updated_at=now()::timestamp`)
 	writeJSON(w, http.StatusOK, map[string]bool{"paused": true})
 }
 
 func (s *Server) handleResumeQueue(w http.ResponseWriter, r *http.Request) {
 	_, _ = s.db.Exec(
-		`INSERT INTO system_flags (key, value, updated_at) VALUES ('queue_paused', 'false', now()::timestamp)
+		`INSERT INTO system_flags (key, value, updated_at) VALUES ('queue_paused', '0', now()::timestamp) ON CONFLICT (key) DO UPDATE SET value='0', updated_at=now()::timestamp
 		 ON CONFLICT (key) DO UPDATE SET value='false', updated_at=now()::timestamp`)
 	writeJSON(w, http.StatusOK, map[string]bool{"paused": false})
 }

server/main.go

@@ -157,12 +157,12 @@ func main() {
 	mux.HandleFunc("POST /sensemaking/sessions", s.requireAuth(s.handleCreateSenseSession))
 	mux.HandleFunc("GET /sensemaking/sessions", s.requireAuth(s.handleListSenseSessions))
 	mux.HandleFunc("GET /sensemaking/sessions/{session_id}", s.requireAuth(s.handleGetSenseSession))
-	mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act1", s.requireAuth(s.handleUpdateAct))
-	mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act2", s.requireAuth(s.handleUpdateAct))
-	mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act3", s.requireAuth(s.handleUpdateAct))
-	mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act1/generate", s.requireAuth(s.handleGenerateAct))
-	mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act2/generate", s.requireAuth(s.handleGenerateAct))
-	mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act3/generate", s.requireAuth(s.handleGenerateAct))
+	mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act1", s.requireAuth(s.makeActHandler("act1")))
+	mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act2", s.requireAuth(s.makeActHandler("act2")))
+	mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act3", s.requireAuth(s.makeActHandler("act3")))
+	mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act1/generate", s.requireAuth(s.makeActGenerateHandler("act1")))
+	mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act2/generate", s.requireAuth(s.makeActGenerateHandler("act2")))
+	mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act3/generate", s.requireAuth(s.makeActGenerateHandler("act3")))
 	mux.HandleFunc("POST /sensemaking/interactions", s.requireAuth(s.handleCreateInteraction))
 	mux.HandleFunc("POST /citations/sync/incremental", s.requireAuth(s.handleSyncCitationsIncremental))
 	mux.HandleFunc("POST /citations/sync/topic/{topic_id}", s.requireAuth(s.handleSyncCitationsTopic))