ingest-hash-trigger SDD task 4：觸發鏈 Gitea webhook → collector sync 直打 named-webhook

- collector Go：trigger.go＋sync 子命令（scan→R2 upload→POST collector-trigger.v1 →2xx 才 MarkIngested 回寫；失敗不回寫＝下輪自然重試；上傳失敗事件不送、同路徑 renamed 連坐不回寫；防呆警告輪照送 warnings）。go test 19/19 綠（httptest mock）。 - workflow v3：rag-ingest-cards.local.yaml 改吃 collector payload（events[] 取代 commits[]；原稿從 R2 撈 CF REST API objects/{r2_key}；新增 renamed 分支＝只更新路徑映射 R5，跨界搬移轉 added/removed；removed 沿用 G9 雙判準；M3 recompute 尾鏈原樣保留，fetch_index 改 R2）；rag-ingest.yaml v1 同步等價改。source_uri 遷移 gitea://→kb://。schema_version!==1 整輪 no-op。 - push-demo-workflow.sh：補 __CF_ACCOUNT_ID__/__R2_BUCKET__/__CF_R2_TOKEN__ 佔位； GITEA_* 標 deprecated-for-ingest（rag-extract* 仍用）。 - Gitea 退場（本 repo 範圍）：legacy JS collector 六檔刪除（git 記錄可查）； install.sh 步驟 5 降級 deprecated；README 落「Gitea 退場切換 checklist」（推 v3→collector sync 實測→拆 Gitea webhook→觀察期＋回滾法）。 - 驗證：yaml parse（raw＋sed 模擬）／flow-config 對齊／JS 語法 node --check／ demo cypher 唯讀 /cypher/search 編譯 missing=[]、出邊序如設計。未部署、未實跑 e2e（推 demo＝leo 閘）。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-19 21:58:39 +08:00
parent fdb7a67484
commit 1c2fa6b294
10 changed files with 495 additions and 581 deletions
@@ -6,6 +6,8 @@
 //	    [--max-removed-ratio 0.4] [--dry-run]
 //	collector upload --root <知識資料夾> --manifest <manifest.json 路徑> \
 //	    [--max-removed-ratio 0.4] [--dry-run]
+//	collector sync   --root <知識資料夾> --manifest <manifest.json 路徑> \
+//	    [--max-removed-ratio 0.4] [--dry-run]
 //
 // scan：走訪 root、對照 manifest、把事件（符合 schemas/collector-trigger.v1.schema.json）
 // 以 JSON 輸出到 stdout，並更新 manifest（--dry-run 不寫）。
@@ -15,7 +17,12 @@
 // R2_BUCKET（絕不落 repo）；--dry-run 只列出會上傳的 key（planned），不碰網路不寫 manifest。
 // 任一上傳失敗＝exit 1（manifest 照存：content_hash 反映現況、ingested_hash 不動＝可重試）。
 //
-// daemon 常駐（launchd）/打 named-webhook 觸發 ingest＝之後的 task（SDD task 4、產品化段）。
+// sync（task 4）＝scan＋upload＋把整輪 payload POST 到 ARCRUN_TRIGGER_URL（arcrun
+// named-webhook 觸發 ingest cypher workflow）。HTTP 2xx 才對本輪送出的 added/modified/
+// renamed 事件回寫 Manifest.MarkIngested；失敗不回寫（下輪自然重試）。上傳失敗的事件
+// 不隨 payload 送出（r2_key 語意＝原稿已在 R2）。防呆警告輪照送 warnings、不含下架事件。
+//
+// daemon 常駐（launchd）＝之後的 task（產品化段 task 11）。
 package main

 import (
@@ -24,8 +31,14 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"time"
 )

+type runMode struct {
+	withUpload  bool
+	withTrigger bool
+}
+
 func main() {
 	if len(os.Args) < 2 {
 		usage()
@@ -33,9 +46,11 @@ func main() {
 	}
 	switch os.Args[1] {
 	case "scan":
-		os.Exit(run(os.Args[2:], false))
+		os.Exit(run(os.Args[2:], runMode{}))
 	case "upload":
-		os.Exit(run(os.Args[2:], true))
+		os.Exit(run(os.Args[2:], runMode{withUpload: true}))
+	case "sync":
+		os.Exit(run(os.Args[2:], runMode{withUpload: true, withTrigger: true}))
 	default:
 		usage()
 		os.Exit(2)
@@ -46,16 +61,18 @@ func usage() {
 	fmt.Fprintln(os.Stderr, `用法:
  collector scan   --root <dir> --manifest <file> [--max-removed-ratio 0.4] [--dry-run]
  collector upload --root <dir> --manifest <file> [--max-removed-ratio 0.4] [--dry-run]
-upload 需環境變數: CF_ACCOUNT_ID / CF_API_TOKEN / R2_BUCKET`)
+  collector sync   --root <dir> --manifest <file> [--max-removed-ratio 0.4] [--dry-run]
+upload 需環境變數: CF_ACCOUNT_ID / CF_API_TOKEN / R2_BUCKET
+sync   另需:      ARCRUN_TRIGGER_URL（named-webhook 觸發完整 URL）`)
 }

-// run 是 scan/upload 共用主體；withUpload＝true 時掃描後把 added/modified 原稿上傳 R2。
-func run(args []string, withUpload bool) int {
+// run 是 scan/upload/sync 共用主體。
+func run(args []string, mode runMode) int {
 	fs := flag.NewFlagSet("collector", flag.ExitOnError)
 	root := fs.String("root", "", "知識資料夾根路徑（必填）")
 	manifestPath := fs.String("manifest", "", "manifest JSON 檔路徑（必填；不存在會建新）")
 	ratio := fs.Float64("max-removed-ratio", DefaultMaxRemovedRatio, "大量刪除防呆門檻（removed 數 > manifest 條目 × 本值 → 全部不下架、只發警告）")
-	dryRun := fs.Bool("dry-run", false, "只輸出事件（upload 模式另列 planned 上傳清單），不更新 manifest、不碰網路")
+	dryRun := fs.Bool("dry-run", false, "只輸出事件（upload/sync 模式另列 planned 清單），不更新 manifest、不碰網路")
 	if err := fs.Parse(args); err != nil {
 		return 2
 	}
@@ -64,15 +81,26 @@ func run(args []string, withUpload bool) int {
 		return 2
 	}

-	// upload 模式先驗設定（fail fast：缺 env 連掃都不掃，不留半套狀態）。
+	// 先驗設定（fail fast：缺 env 連掃都不掃，不留半套狀態）。
 	var client *R2Client
-	if withUpload && !*dryRun {
-		cfg, err := LoadR2ConfigFromEnv()
-		if err != nil {
-			fmt.Fprintln(os.Stderr, "collector:", err)
-			return 2
+	var triggerURL string
+	if !*dryRun {
+		if mode.withUpload {
+			cfg, err := LoadR2ConfigFromEnv()
+			if err != nil {
+				fmt.Fprintln(os.Stderr, "collector:", err)
+				return 2
+			}
+			client = NewR2Client(cfg)
+		}
+		if mode.withTrigger {
+			u, err := LoadTriggerURLFromEnv()
+			if err != nil {
+				fmt.Fprintln(os.Stderr, "collector:", err)
+				return 2
+			}
+			triggerURL = u
 		}
-		client = NewR2Client(cfg)
 	}

 	absRoot, err := filepath.Abs(*root)
@@ -96,8 +124,10 @@ func run(args []string, withUpload bool) int {
 		return fail(err)
 	}

+	exitCode := 0
+
 	var uploads []UploadResult
-	if withUpload {
+	if mode.withUpload {
 		if *dryRun {
 			uploads = []UploadResult{}
 			for _, ev := range payload.Events {
@@ -107,6 +137,38 @@ func run(args []string, withUpload bool) int {
 			}
 		} else {
 			uploads = UploadChanged(absRoot, payload.Events, client)
+			for _, r := range uploads {
+				if r.Status == "failed" {
+					exitCode = 1 // 有敗＝非零退出；ingested_hash 未動＝下輪自然重試
+				}
+			}
+		}
+	}
+
+	// sync：POST 觸發 → 2xx 才回寫 MarkIngested（在 m.Save 之前，回寫才進得了檔）。
+	outPayload := payload
+	var dispatch *TriggerResult
+	if mode.withTrigger {
+		if *dryRun {
+			dispatch = &TriggerResult{Status: "planned"}
+		} else {
+			sendable, dropped := BuildSendablePayload(payload, uploads)
+			outPayload = sendable
+			dispatch = &TriggerResult{DroppedPaths: dropped}
+			if len(sendable.Events) == 0 && len(sendable.Warnings) == 0 {
+				dispatch.Status = "skipped_no_changes" // 無變更輪不發送（schema 註明空發也合法，但沒必要）
+			} else {
+				status, terr := SendTrigger(triggerURL, sendable, nil)
+				dispatch.HTTPStatus = status
+				if terr != nil {
+					dispatch.Status = "failed"
+					dispatch.Error = terr.Error()
+					exitCode = 1
+				} else {
+					dispatch.Status = "sent"
+					dispatch.MarkedCount = MarkIngestedEvents(m, sendable.Events, dropped, time.Now().Unix())
+				}
+			}
 		}
 	}

@@ -116,18 +178,18 @@ func run(args []string, withUpload bool) int {
 		}
 	}

-	exitCode := 0
-	var out any = payload
-	if withUpload {
+	var out any = outPayload
+	if mode.withTrigger {
+		out = struct {
+			Trigger  *TriggerPayload `json:"trigger"`
+			Uploads  []UploadResult  `json:"uploads"`
+			Dispatch *TriggerResult  `json:"dispatch"`
+		}{outPayload, uploads, dispatch}
+	} else if mode.withUpload {
 		out = struct {
 			Trigger *TriggerPayload `json:"trigger"`
 			Uploads []UploadResult  `json:"uploads"`
-		}{payload, uploads}
-		for _, r := range uploads {
-			if r.Status == "failed" {
-				exitCode = 1 // 有敗＝非零退出；manifest 已存、ingested_hash 未動＝下輪自然重試
-			}
-		}
+		}{outPayload, uploads}
 	}
 	data, err := json.MarshalIndent(out, "", "  ")
 	if err != nil {