diff --git a/README.md b/README.md index 405698f..98f12f3 100644 --- a/README.md +++ b/README.md @@ -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 [--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 上做。 diff --git a/direct.go b/direct.go new file mode 100644 index 0000000..551b324 --- /dev/null +++ b/direct.go @@ -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 [--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..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 +} diff --git a/main.go b/main.go index 8476478..f0ab7a3 100644 --- a/main.go +++ b/main.go @@ -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 --manifest [--max-removed-ratio 0.4] [--dry-run] collector upload --root --manifest [--max-removed-ratio 0.4] [--dry-run] collector sync --root --manifest [--max-removed-ratio 0.4] [--dry-run] + collector direct --config [--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 共用主體。