feat(daemon): direct 直送萃取同步模式（無 R2/Gitea，youlin live e2e 綠）

SDD ingest-hash-trigger task 11：daemon 跨平台核心 + 產品承諾核心實證（丟檔進資料夾 → AI 查得到）。 - collector 新增 direct 子命令（direct.go）：監看資料夾 → hash 差異偵測（沿用 Scan） → 讀檔內容 inline POST 進實例 rag_ingest_direct workflow；刪檔 → POST {page_name,path} 進 rag_takedown_direct。設定走 --config JSON（install/direct-config.sample.json）。常駐輪詢＝純 stdlib ticker，零 CGo，darwin/arm64 + windows/amd64 交叉編譯 OK。 - workflows/rag-ingest-direct：ask_llm（rag_extract_one）+ parse_card/post_block/post_triplet （rag_ingest）拼成單一自足直鏈，繞開 R2 撈原稿與 Gitea 落卡。blocks/triplets 寫法與 rag_ingest 逐欄一致 → rag_chat 檢索得到。 - workflows/rag-takedown-direct：不含 __CARDS_PREFIX__ 閘的下架（rag_ingest removed 分支的前綴閘會擋掉資料夾根的 direct 檔——2026-07-20 live e2e 實撞）。 dogfooding（D29 daemon 薄殼豁免）守住：daemon 只監看/讀檔/算 hash/HTTP POST，萃取/切塊/RAG 全在 Arcrun workflow。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Xabu4T28MmQEiYDwT2KnKW
2026-07-20 11:15:21 +00:00
parent 1c2fa6b294
commit ad34a87210
3 changed files with 339 additions and 1 deletions
@@ -86,3 +86,42 @@ design 鐵律段明言保留；被刪掉的是「Gitea push 事件」這個來
 測試：`go test ./...` 19/19——掃描 7＋上傳 7＋sync/trigger 5（成功回寫＋無變更輪不重發／
 失敗不回寫＋修復後重試成功／防呆警告輪照送 warnings 零事件／上傳失敗事件擋下＋renamed
 連坐不回寫／URL env 驗證），全部 httptest mock（**未實跑雲端 e2e**——等切換日與 leo 一起驗）。
+
+## `direct`：daemon 直送萃取、無 R2 同步模式（SDD task 11）
+
+`direct`＝**產品承諾核心的最短路徑**：監看資料夾 → 偵測新增/改動檔（沿用 `Scan` 的 hash 差異
+偵測，與 `sync` 同一套）→ 讀檔內容 **inline POST** 進實例的 `rag_ingest_direct` workflow
+（LLM 萃卡 → 機械切塊 → 寫 kbdb，**全在 Arcrun workflow**）。刪檔 → POST `{page_name, path}`
+進 `rag_takedown_direct`（按 page_name 標 kbdb blocks/triplets `deprecated`）。
+
+**繞開 R2/Gitea**：既有 `sync` 走 collector→R2→`rag_ingest`（要 R2 bucket＝綁卡），`rag_extract_one`
+又要 Gitea repo 落卡。`direct` 兩者都不要——落地「丟檔進資料夾 → AI 查得到」的零綁卡版。
+
+```
+collector direct --config <config.json> [--once] [--dry-run]
+```
+
+- `--once`：掃一輪即退出（測試/cron）；預設常駐輪詢（`poll_interval_sec`，純 stdlib ticker，跨平台）。
+- 設定檔（JSON，見 `install/direct-config.sample.json`）：`watch_folder` / `manifest` /
+  `cypher_url` / `namespace` / `api_key`（空＝namespace）/ `library`（空＝kb）/
+  `poll_interval_sec`（空＝5）/ `max_removed_ratio`（空＝0.4）/ `ingest_workflow`
+  （空＝`rag_ingest_direct`）/ `removed_workflow`（空＝`rag_takedown_direct`）。
+- **dogfooding（D29 daemon 薄殼豁免）**：本模式只「監看／讀檔／算 hash／HTTP POST」原生 Go——
+  萃取/切塊/RAG 一律在實例 workflow，daemon 內零 LLM/切塊邏輯。
+- **回寫語意**：POST 回 2xx 才 `Manifest.MarkIngested`（下輪不重送）；非 2xx 不回寫＝下輪重試。
+- **大量刪除防呆**：沿用 `Scan` 的 `mass_delete_guard`（removed > manifest×`max_removed_ratio`
+  → 本輪不下架、只回報警告）。
+
+**配套 workflow**（`workflows/rag-ingest-direct.local.yaml` / `rag-takedown-direct.local.yaml`；
+用 `install/push-demo-workflow.sh` 推，env 指向目標實例的 CYPHER/KBDB/HTTPREQ/CODE/LLM_MODEL/LIBRARY）：
+blocks/triplets 寫法與 `rag_ingest` 逐欄一致 → `rag_chat` 一視同仁檢索得到。
+> ⚠️ 下架另立 `rag_takedown_direct` 而非重用 `rag_ingest` removed 分支：後者在 `collect_changed`
+> 有 `__CARDS_PREFIX__` 閘，direct 模式的檔在資料夾根會被擋掉零下架（2026-07-20 live e2e 實撞）。
+
+**youlin live e2e 實證（2026-07-20，非 mock）**：
+- 丟 `請假規則.md`/`差旅政策.md` → daemon `ingested`(200) → `rag_chat` 帶出處正確作答。
+- 刪 `差旅政策.md` → daemon `removed`(200) → `rag_chat`「知識庫裡沒有這方面資料」（blocks 全 deprecated）。
+
+**跨平台**：純 stdlib、零 CGo、輪詢式偵測（不依賴 fsnotify）→ `GOOS=darwin GOARCH=arm64` /
+`GOOS=windows GOARCH=amd64` 直接交叉編譯（各約 5.6M / 5.9M）。托盤殼（Wails）＋Mac `.app`
+簽章/TCC 是下一刀，且 Mac 打包要在 Mac 上做。
@@ -0,0 +1,290 @@
+// direct.go — daemon「直送萃取、無 R2」同步模式（SDD ingest-hash-trigger task 11）。
+//
+// 既有 sync 走 collector → R2 → rag_ingest（需 R2 bucket＝綁卡）。direct 模式繞開 R2/Gitea：
+// 監看資料夾 → 偵測新增/改動檔（沿用 Scan 的 hash 差異偵測）→ 讀檔內容 inline POST 進實例的
+// rag_ingest_direct workflow（LLM 萃卡 → 機械切塊 → 寫 kbdb，全在 Arcrun workflow 裡完成）。
+// 刪檔 → 把 removed 事件（collector-trigger.v1）POST 進實例的 rag_ingest workflow removed 分支
+// （只按 page_name 讀 kbdb blocks 並標 deprecated，不碰 R2）。
+//
+// dogfooding（D29 daemon 薄殼豁免）：本檔只做「監看／讀檔／算 hash／HTTP POST」——原生 Go。
+// 萃取／切塊／RAG 一律在實例 workflow，daemon 內零 LLM/切塊邏輯。
+//
+// 用法：
+//
+//	collector direct --config <config.json> [--once] [--dry-run]
+//
+// --once：掃一輪就退出（測試/cron 用）；預設常駐輪詢（poll_interval_sec）。
+// --dry-run：只列出會送出的動作，不 POST、不寫 manifest。
+//
+// 跨平台：純 stdlib、輪詢式偵測（不依賴 fsnotify）＝零 CGo，darwin/arm64、windows/amd64 直接交叉編譯。
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+// DirectConfig 是 direct 模式的設定檔（JSON）。設定只走檔案/環境，不落 code。
+type DirectConfig struct {
+	WatchFolder string `json:"watch_folder"`          // 監看的知識資料夾（必填）
+	Manifest    string `json:"manifest"`              // manifest JSON 路徑（必填；不存在會建新）
+	CypherURL   string `json:"cypher_url"`            // 實例 cypher base（必填），如 https://arcrun-cypher-executor.<acct>.workers.dev
+	Namespace   string `json:"namespace"`             // 租戶 namespace（必填），如 demo
+	APIKey      string `json:"api_key"`               // X-Arcrun-API-Key（空＝沿用 namespace，demo 慣例）
+	Library     string `json:"library"`               // 藏書地圖歸庫鍵（空＝kb）
+	IngestWF    string `json:"ingest_workflow"`       // 直送萃取 workflow 名（空＝rag_ingest_direct）
+	RemovedWF   string `json:"removed_workflow"`      // 下架 workflow 名（空＝rag_takedown_direct；吃 {page_name,path}）
+	PollSec     int    `json:"poll_interval_sec"`     // 輪詢間隔秒（空/0＝5）
+	MaxRemoved  float64 `json:"max_removed_ratio"`    // 大量刪除防呆門檻（空/0＝0.4）
+}
+
+// LoadDirectConfig 讀設定檔並補預設值 + 基本驗證。
+func LoadDirectConfig(path string) (*DirectConfig, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("讀 config 失敗：%w", err)
+	}
+	var c DirectConfig
+	if err := json.Unmarshal(data, &c); err != nil {
+		return nil, fmt.Errorf("config JSON 解析失敗：%w", err)
+	}
+	var missing []string
+	if c.WatchFolder == "" {
+		missing = append(missing, "watch_folder")
+	}
+	if c.Manifest == "" {
+		missing = append(missing, "manifest")
+	}
+	if c.CypherURL == "" {
+		missing = append(missing, "cypher_url")
+	}
+	if c.Namespace == "" {
+		missing = append(missing, "namespace")
+	}
+	if len(missing) > 0 {
+		return nil, fmt.Errorf("config 缺必填欄位：%s", strings.Join(missing, ", "))
+	}
+	if c.APIKey == "" {
+		c.APIKey = c.Namespace
+	}
+	if c.Library == "" {
+		c.Library = "kb"
+	}
+	if c.IngestWF == "" {
+		c.IngestWF = "rag_ingest_direct"
+	}
+	if c.RemovedWF == "" {
+		c.RemovedWF = "rag_takedown_direct"
+	}
+	if c.PollSec <= 0 {
+		c.PollSec = 5
+	}
+	if c.MaxRemoved <= 0 {
+		c.MaxRemoved = DefaultMaxRemovedRatio
+	}
+	c.CypherURL = strings.TrimSuffix(c.CypherURL, "/")
+	return &c, nil
+}
+
+// directHTTP 是 direct 模式共用的 HTTP client（萃取 workflow 可能同步跑 LLM，放寬 timeout）。
+var directHTTP = &http.Client{Timeout: 300 * time.Second}
+
+// triggerURL 組出 named-webhook 觸發完整 URL。
+func (c *DirectConfig) triggerURL(workflow string) string {
+	return fmt.Sprintf("%s/webhooks/named/%s/%s/trigger", c.CypherURL, c.Namespace, workflow)
+}
+
+// postJSON POST 一個 JSON body 到 url，回傳 HTTP 狀態碼與回應片段。非 2xx 視為錯誤。
+func (c *DirectConfig) postJSON(url string, body any) (int, string, error) {
+	data, err := json.Marshal(body)
+	if err != nil {
+		return 0, "", err
+	}
+	req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(data))
+	if err != nil {
+		return 0, "", err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Arcrun-API-Key", c.APIKey)
+	resp, err := directHTTP.Do(req)
+	if err != nil {
+		return 0, "", err
+	}
+	defer resp.Body.Close()
+	snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return resp.StatusCode, string(snippet), fmt.Errorf("HTTP %d：%s", resp.StatusCode, strings.TrimSpace(string(snippet)))
+	}
+	return resp.StatusCode, string(snippet), nil
+}
+
+// pageNameOf 從相對路徑導出頁名（basename 去副檔名），與 rag_ingest collect_changed 的 pageOf 同語意。
+func pageNameOf(relPath string) string {
+	base := relPath
+	if i := strings.LastIndex(base, "/"); i >= 0 {
+		base = base[i+1:]
+	}
+	return strings.TrimSuffix(base, filepath.Ext(base))
+}
+
+// DirectResult 是單一事件的直送結果（隨每輪 log 輸出）。
+type DirectResult struct {
+	Type       string `json:"type"`
+	Path       string `json:"path"`
+	Status     string `json:"status"` // ingested | removed | planned | failed | skipped
+	HTTPStatus int    `json:"http_status,omitempty"`
+	Error      string `json:"error,omitempty"`
+}
+
+// RunDirectOnce 掃一輪、直送 added/modified/renamed、下架 removed，並在 2xx 後回寫 manifest。
+// 回傳本輪結果清單與退出碼建議（有失敗＝1）。
+func RunDirectOnce(cfg *DirectConfig, dryRun bool) ([]DirectResult, int, *TriggerPayload) {
+	results := []DirectResult{}
+	exit := 0
+
+	absRoot, err := filepath.Abs(cfg.WatchFolder)
+	if err != nil {
+		return append(results, DirectResult{Status: "failed", Error: err.Error()}), 1, nil
+	}
+	absManifest, err := filepath.Abs(cfg.Manifest)
+	if err != nil {
+		return append(results, DirectResult{Status: "failed", Error: err.Error()}), 1, nil
+	}
+	m, err := LoadManifest(absManifest, absRoot)
+	if err != nil {
+		return append(results, DirectResult{Status: "failed", Error: err.Error()}), 1, nil
+	}
+	payload, err := Scan(absRoot, m, ScanOptions{
+		MaxRemovedRatio: cfg.MaxRemoved,
+		SkipPaths:       map[string]bool{absManifest: true},
+	})
+	if err != nil {
+		return append(results, DirectResult{Status: "failed", Error: err.Error()}), 1, nil
+	}
+
+	now := time.Now().Unix()
+	for _, ev := range payload.Events {
+		switch ev.Type {
+		case "added", "modified", "renamed":
+			// renamed 在 direct 模式視同 added：內容未變但為求 kbdb 有這頁名的卡，重送一次萃取
+			//（頁名可能改變＝要新頁名的卡）。冪等由 kbdb 端承擔（同頁名覆蓋語意）。
+			res := DirectResult{Type: ev.Type, Path: ev.Path}
+			full := filepath.Join(absRoot, filepath.FromSlash(ev.Path))
+			content, rerr := os.ReadFile(full)
+			if rerr != nil {
+				res.Status, res.Error = "failed", "讀檔失敗："+rerr.Error()
+				results = append(results, res)
+				exit = 1
+				continue
+			}
+			if dryRun {
+				res.Status = "planned"
+				results = append(results, res)
+				continue
+			}
+			status, _, perr := cfg.postJSON(cfg.triggerURL(cfg.IngestWF), map[string]any{
+				"page_name": pageNameOf(ev.Path),
+				"path":      ev.Path,
+				"content":   string(content),
+				"library":   cfg.Library,
+			})
+			res.HTTPStatus = status
+			if perr != nil {
+				res.Status, res.Error = "failed", perr.Error()
+				exit = 1
+			} else {
+				res.Status = "ingested"
+				m.MarkIngested(ev.Path, ev.SourceHash, now) // 2xx 才回寫（下輪不重送）
+			}
+			results = append(results, res)
+
+		case "removed":
+			res := DirectResult{Type: ev.Type, Path: ev.Path}
+			if dryRun {
+				res.Status = "planned"
+				results = append(results, res)
+				continue
+			}
+			// 下架＝POST {page_name, path} 進 rag_takedown_direct（按 page_name 讀 kbdb blocks
+			// 標 deprecated，不碰 R2；獨立於 rag_ingest 的 __CARDS_PREFIX__ 閘——direct 模式檔在
+			// 資料夾根，會被 rag_ingest 的前綴閘擋掉，故自帶不含前綴閘的下架 workflow）。
+			status, _, perr := cfg.postJSON(cfg.triggerURL(cfg.RemovedWF), map[string]any{
+				"page_name": pageNameOf(ev.Path),
+				"path":      ev.Path,
+			})
+			res.HTTPStatus = status
+			if perr != nil {
+				res.Status, res.Error = "failed", perr.Error()
+				exit = 1
+			} else {
+				res.Status = "removed"
+			}
+			results = append(results, res)
+		}
+	}
+
+	// 防呆警告輪：Scan 已壓下 removed 事件，這裡只回報警告不下架。
+	for _, w := range payload.Warnings {
+		results = append(results, DirectResult{Type: "warning", Status: "skipped", Error: w.Code + ": " + w.Message})
+	}
+
+	if !dryRun {
+		if err := m.Save(absManifest); err != nil {
+			results = append(results, DirectResult{Status: "failed", Error: "manifest 存檔失敗：" + err.Error()})
+			exit = 1
+		}
+	}
+	return results, exit, payload
+}
+
+// runDirect 是 `collector direct` 子命令主體。
+func runDirect(args []string) int {
+	fs := newFlagSet()
+	configPath := fs.String("config", "", "direct 模式設定檔（JSON）路徑（必填）")
+	once := fs.Bool("once", false, "掃一輪即退出（測試/cron；預設常駐輪詢）")
+	dryRun := fs.Bool("dry-run", false, "只列出會送出的動作，不 POST、不寫 manifest")
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	if *configPath == "" {
+		fmt.Fprintln(os.Stderr, "錯誤：--config 為必填")
+		return 2
+	}
+	cfg, err := LoadDirectConfig(*configPath)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, "collector direct:", err)
+		return 2
+	}
+
+	runOne := func() int {
+		results, exit, _ := RunDirectOnce(cfg, *dryRun)
+		out, _ := json.MarshalIndent(struct {
+			At      string         `json:"at"`
+			Folder  string         `json:"folder"`
+			Results []DirectResult `json:"results"`
+		}{time.Now().Format(time.RFC3339), cfg.WatchFolder, results}, "", "  ")
+		fmt.Println(string(out))
+		return exit
+	}
+
+	if *once {
+		return runOne()
+	}
+	// 常駐輪詢：純 stdlib ticker，跨平台。首輪立即跑。
+	fmt.Fprintf(os.Stderr, "collector direct daemon 啟動：監看 %s → %s（每 %ds 掃一輪）\n",
+		cfg.WatchFolder, cfg.triggerURL(cfg.IngestWF), cfg.PollSec)
+	runOne()
+	ticker := time.NewTicker(time.Duration(cfg.PollSec) * time.Second)
+	defer ticker.Stop()
+	for range ticker.C {
+		runOne()
+	}
+	return 0
+}
@@ -51,19 +51,28 @@ func main() {
 		os.Exit(run(os.Args[2:], runMode{withUpload: true}))
 	case "sync":
 		os.Exit(run(os.Args[2:], runMode{withUpload: true, withTrigger: true}))
+	case "direct":
+		os.Exit(runDirect(os.Args[2:]))
 	default:
 		usage()
 		os.Exit(2)
 	}
 }

+// newFlagSet 是子命令共用的 flag.FlagSet 建構子（ExitOnError＝解析失敗直接退出）。
+func newFlagSet() *flag.FlagSet {
+	return flag.NewFlagSet("collector", flag.ExitOnError)
+}
+
 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]
  collector sync   --root <dir> --manifest <file> [--max-removed-ratio 0.4] [--dry-run]
+  collector direct --config <config.json> [--once] [--dry-run]
 upload 需環境變數: CF_ACCOUNT_ID / CF_API_TOKEN / R2_BUCKET
-sync   另需:      ARCRUN_TRIGGER_URL（named-webhook 觸發完整 URL）`)
+sync   另需:      ARCRUN_TRIGGER_URL（named-webhook 觸發完整 URL）
+direct（無 R2）:  監看資料夾 → 讀檔內容直送實例 rag_ingest_direct workflow；設定走 --config JSON`)
 }

 // run 是 scan/upload/sync 共用主體。