diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 549ee797..c625fb0a 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -249,6 +249,10 @@ imports: sysadmin: - config controlcenter/config: [] + controlcenter/embed: [] + controlcenter/docs: + - controlcenter/embed + - controlcenter/store controlcenter/pipeline: [] controlcenter/store: [] controlcenter/tasks: @@ -256,6 +260,8 @@ imports: - controlcenter/store controlcenter/server: - controlcenter/config + - controlcenter/docs + - controlcenter/embed - controlcenter/pipeline - controlcenter/store - controlcenter/tasks diff --git a/cmd/control-center/main.go b/cmd/control-center/main.go index 77f2dfa7..e9e3ce75 100644 --- a/cmd/control-center/main.go +++ b/cmd/control-center/main.go @@ -16,6 +16,8 @@ import ( "time" "chat-agent-system/internal/controlcenter/config" + "chat-agent-system/internal/controlcenter/docs" + "chat-agent-system/internal/controlcenter/embed" "chat-agent-system/internal/controlcenter/server" "chat-agent-system/internal/controlcenter/store" ) @@ -77,6 +79,23 @@ func cmdServe(args []string) error { } srv := server.New(cfg, st) + var emb *embed.HTTPClient + if cfg.EmbeddingEndpoint != "" { + emb = embed.NewHTTP(cfg.EmbeddingEndpoint, cfg.EmbeddingAPIKey, cfg.EmbeddingModel, embed.Dim) + } else { + slog.Warn("embedding endpoint not configured; search runs keyword-only") + } + syncer := docs.New(st, cfg.RepoRoot, emb) + var embIface embed.Embedder + if emb != nil { + embIface = emb + } + srv.SetDocs(embIface, syncer) + + syncCtx, stopSync := context.WithCancel(ctx) + defer stopSync() + syncer.StartPeriodic(syncCtx, time.Duration(cfg.DocSyncSecond)*time.Second) + reapCtx, stopReaper := context.WithCancel(ctx) defer stopReaper() srv.StartLeaseReaper(reapCtx) @@ -86,6 +105,7 @@ func cmdServe(args []string) error { sig := make(chan os.Signal, 1) signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) <-sig + stopSync() stopReaper() shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() diff --git a/internal/controlcenter/config/config.go b/internal/controlcenter/config/config.go index deadaf95..8d596f20 100644 --- a/internal/controlcenter/config/config.go +++ b/internal/controlcenter/config/config.go @@ -8,6 +8,7 @@ import ( "fmt" "os" "path/filepath" + "strconv" "strings" ) @@ -79,11 +80,20 @@ func Load() (Config, error) { AuditEnabled: getenv("CC_AUDIT_ENABLED", "true") == "true", AuditDailyAt: getenv("CC_AUDIT_DAILY_AT", "03:30"), - DocSyncSecond: 60, + DocSyncSecond: getint("CC_DOC_SYNC_SECONDS", 60), } return cfg, nil } +func getint(key string, def int) int { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return def +} + func getenv(key, def string) string { if v := os.Getenv(key); v != "" { return v diff --git a/internal/controlcenter/docs/chunk.go b/internal/controlcenter/docs/chunk.go new file mode 100644 index 00000000..3c5177ee --- /dev/null +++ b/internal/controlcenter/docs/chunk.go @@ -0,0 +1,136 @@ +package docs + +import ( + "strings" +) + +// maxChunkTokens bounds one chunk; oversized sections are bisected by +// paragraph until they fit (or cannot be split further). +const maxChunkTokens = 1800 + +// Chunk is one embeddable piece of a document. +type Chunk struct { + Seq int + Heading string + Content string +} + +// chunkMarkdown splits md at ATX heading lines; content before the first +// heading becomes a chunk with an empty heading. Oversized sections are then +// bisected by paragraph. Heading text is kept inside the content so the +// embedding carries it. +func chunkMarkdown(md string) []Chunk { + var chunks []Chunk + for _, sec := range splitSections(md) { + chunks = append(chunks, bisect(sec.heading, sec.content)...) + } + for i := range chunks { + chunks[i].Seq = i + } + return chunks +} + +type section struct { + heading string + content string +} + +func splitSections(md string) []section { + var secs []section + var cur section + var lines []string + flush := func() { + body := strings.TrimSpace(strings.Join(lines, "\n")) + if cur.heading == "" && body == "" { + lines = nil + return + } + cur.content = strings.TrimSpace(cur.heading + "\n" + body) + if cur.content != "" { + secs = append(secs, cur) + } + lines = nil + } + for _, line := range strings.Split(md, "\n") { + if h, ok := atxHeading(line); ok { + flush() + cur = section{heading: h} + continue + } + lines = append(lines, line) + } + flush() + return secs +} + +// atxHeading returns the heading line (with # marks) for ATX headings. +func atxHeading(line string) (string, bool) { + t := strings.TrimSpace(line) + if !strings.HasPrefix(t, "#") { + return "", false + } + i := 0 + for i < len(t) && t[i] == '#' { + i++ + } + if i > 6 || i >= len(t) || t[i] != ' ' { + return "", false + } + return t, true +} + +// bisect splits an oversized section by paragraphs, halving recursively; a +// single oversized paragraph is cut by rune count. +func bisect(heading, content string) []Chunk { + if estimateTokens(content) <= maxChunkTokens { + return []Chunk{{Heading: heading, Content: content}} + } + paras := strings.Split(content, "\n\n") + if len(paras) > 1 { + mid := len(paras) / 2 + left := bisect(heading, strings.TrimSpace(strings.Join(paras[:mid], "\n\n"))) + right := bisect(heading, strings.TrimSpace(strings.Join(paras[mid:], "\n\n"))) + return append(left, right...) + } + // single huge paragraph: hard cut on a rune boundary + runes := []rune(content) + half := len(runes) / 2 + for half > 0 && half < len(runes)-1 && runes[half] != ' ' && runes[half] != '\n' && runes[half] != '。' { + half++ + } + if half <= 0 || half >= len(runes) { + half = len(runes) / 2 + } + left := bisect(heading, strings.TrimSpace(string(runes[:half]))) + right := bisect(heading, strings.TrimSpace(string(runes[half:]))) + return append(left, right...) +} + +// estimateTokens approximates token count: CJK runes cost 2, ASCII runs cost +// ceil(len/4). Same heuristic as internal/kb. +func estimateTokens(s string) int { + s = strings.TrimSpace(s) + if s == "" { + return 1 + } + tokens, asciiRun := 0, 0 + flush := func() { + if asciiRun > 0 { + tokens += (asciiRun + 3) / 4 + asciiRun = 0 + } + } + for _, r := range s { + if r <= 127 { + asciiRun++ + continue + } + flush() + tokens += 2 + } + flush() + if tokens < 1 { + return 1 + } + return tokens +} diff --git a/internal/controlcenter/docs/chunk_test.go b/internal/controlcenter/docs/chunk_test.go new file mode 100644 index 00000000..098f22e7 --- /dev/null +++ b/internal/controlcenter/docs/chunk_test.go @@ -0,0 +1,112 @@ +package docs + +import ( + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestChunkMarkdownHeadings(t *testing.T) { + md := "# Doc Title\n\nintro text\n\n## Section A\n\nalpha body\n\n### Sub A1\n\nsub body\n\n## Section B\n\nbeta body\n" + chunks := chunkMarkdown(md) + require.Len(t, chunks, 4) + assert.Equal(t, "# Doc Title", chunks[0].Heading) + assert.Contains(t, chunks[0].Content, "intro text") + assert.Equal(t, "## Section A", chunks[1].Heading) + assert.Contains(t, chunks[1].Content, "alpha body") + assert.NotContains(t, chunks[1].Content, "sub body") + assert.Equal(t, "### Sub A1", chunks[2].Heading) + assert.Equal(t, "## Section B", chunks[3].Heading) + for i, c := range chunks { + assert.Equal(t, i, c.Seq) + } +} + +func TestChunkMarkdownPreambleOnly(t *testing.T) { + chunks := chunkMarkdown("just some text\nwithout headings\n") + require.Len(t, chunks, 1) + assert.Equal(t, "", chunks[0].Heading) + assert.Contains(t, chunks[0].Content, "just some text") +} + +func TestChunkMarkdownEmpty(t *testing.T) { + assert.Empty(t, chunkMarkdown("")) + assert.Empty(t, chunkMarkdown("\n\n \n")) +} + +func TestChunkMarkdownLongSectionBisected(t *testing.T) { + // Build one section with many paragraphs, well over maxChunkTokens. + var sb strings.Builder + sb.WriteString("## Big\n\n") + for i := 0; i < 60; i++ { + fmt.Fprintf(&sb, "paragraph %d %s\n\n", i, strings.Repeat("word ", 100)) + } + chunks := chunkMarkdown(sb.String()) + require.Greater(t, len(chunks), 1, "oversized section must be split") + for _, c := range chunks { + assert.Equal(t, "## Big", c.Heading) + assert.LessOrEqual(t, estimateTokens(c.Content), maxChunkTokens) + } + // all paragraphs survive the split + joined := strings.Join(func() []string { + var out []string + for _, c := range chunks { + out = append(out, c.Content) + } + return out + }(), "\n") + for i := 0; i < 60; i++ { + assert.Contains(t, joined, fmt.Sprintf("paragraph %d", i)) + } +} + +func TestChunkMarkdownSingleHugeParagraph(t *testing.T) { + md := "## Blob\n\n" + strings.Repeat("汉字", 3000) + "\n" + chunks := chunkMarkdown(md) + require.Greater(t, len(chunks), 1) + for _, c := range chunks { + assert.LessOrEqual(t, estimateTokens(c.Content), maxChunkTokens) + } +} + +func TestChunkHashStable(t *testing.T) { + h1 := chunkHash("doc", ".doc/a.md", 0, "content") + h2 := chunkHash("doc", ".doc/a.md", 0, "content") + assert.Equal(t, h1, h2) + assert.NotEqual(t, h1, chunkHash("doc", ".doc/a.md", 1, "content"), "seq matters") + assert.NotEqual(t, h1, chunkHash("artifact", ".doc/a.md", 0, "content"), "owner kind matters") +} + +func TestSplitFrontMatter(t *testing.T) { + fm, body := splitFrontMatter("---\ntitle: My Doc\nstatus: done\n---\n# Heading\nbody\n") + assert.Equal(t, "My Doc", fm["title"]) + assert.Equal(t, "done", fm["status"]) + assert.True(t, strings.HasPrefix(body, "# Heading")) + + fm, body = splitFrontMatter("# No front matter\n") + assert.Empty(t, fm) + assert.Contains(t, body, "# No front matter") +} + +func TestDocTitleFallback(t *testing.T) { + assert.Equal(t, "FM Title", docTitle(map[string]any{"title": "FM Title"}, "# H1\n", "x.md")) + assert.Equal(t, "H1 Title", docTitle(map[string]any{}, "# H1 Title\nbody\n", "x.md")) + assert.Equal(t, "file-name", docTitle(map[string]any{}, "no heading\n", ".doc/ops/file-name.md")) +} + +func TestTaxonomyOf(t *testing.T) { + assert.Equal(t, "ops", taxonomyOf(".doc/ops/CURRENT.md")) + assert.Equal(t, "ops", taxonomyOf(".doc/ops/short-term/x.md")) + assert.Equal(t, "", taxonomyOf(".doc/README.md")) + assert.Equal(t, "", taxonomyOf("other/x.md")) +} + +func TestEstimateTokens(t *testing.T) { + assert.Equal(t, 1, estimateTokens("")) + assert.Equal(t, 2, estimateTokens("汉")) // CJK rune = 2 + assert.Equal(t, 1, estimateTokens("abcd")) // 4 ascii / 4 + assert.Equal(t, 2, estimateTokens("abcde")) // ceil(5/4) +} diff --git a/internal/controlcenter/docs/docs.go b/internal/controlcenter/docs/docs.go new file mode 100644 index 00000000..d05efee2 --- /dev/null +++ b/internal/controlcenter/docs/docs.go @@ -0,0 +1,376 @@ +// Package docs syncs .doc markdown files into cc_documents/cc_doc_chunks and +// embeds chunk content for pgvector retrieval. Embedding failures degrade +// gracefully: rows are stored with NULL embeddings and the sync succeeds. +package docs + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io/fs" + "log/slog" + "os" + "path/filepath" + "strings" + "time" + + "gopkg.in/yaml.v3" + + "chat-agent-system/internal/controlcenter/embed" + "chat-agent-system/internal/controlcenter/store" +) + +// Syncer ingests repository docs and stage artifacts into the vector store. +type Syncer struct { + st *store.Store + repoRoot string + emb embed.Embedder // nil = embeddings unavailable, chunks stored with NULL +} + +func New(st *store.Store, repoRoot string, emb embed.Embedder) *Syncer { + return &Syncer{st: st, repoRoot: repoRoot, emb: emb} +} + +// Stats summarizes one Sync run. +type Stats struct { + Added int `json:"added"` + Updated int `json:"updated"` + Removed int `json:"removed"` + Unchanged int `json:"unchanged"` + Chunks int `json:"chunks"` + EmbedFailed bool `json:"embed_failed"` +} + +// Sync walks /.doc for *.md files (skipping assets/ directories), +// upserts changed documents, re-chunks them idempotently, and removes rows +// for files that disappeared. +func (s *Syncer) Sync(ctx context.Context) (Stats, error) { + var stats Stats + docRoot := filepath.Join(s.repoRoot, ".doc") + + type diskDoc struct { + rel string // repo-relative slash path, e.g. .doc/ops/CURRENT.md + sha string + full string + } + seen := map[string]diskDoc{} + err := filepath.WalkDir(docRoot, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + if d.Name() == "assets" && path != docRoot { + return filepath.SkipDir + } + return nil + } + if !strings.HasSuffix(d.Name(), ".md") { + return nil + } + rel, err := filepath.Rel(s.repoRoot, path) + if err != nil { + return err + } + rel = filepath.ToSlash(rel) + raw, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read %s: %w", rel, err) + } + seen[rel] = diskDoc{rel: rel, sha: sha256hex(raw), full: path} + return nil + }) + if err != nil { + if os.IsNotExist(err) { + slog.Warn("doc sync: .doc root missing", "root", docRoot) + return stats, nil + } + return stats, err + } + + // Existing rows, to diff against. + existing := map[string]struct { + id string + sha string + }{} + rows, err := s.st.Pool.Query(ctx, `SELECT id, path, sha256 FROM cc_documents`) + if err != nil { + return stats, fmt.Errorf("list cc_documents: %w", err) + } + for rows.Next() { + var id, path, sha string + if err := rows.Scan(&id, &path, &sha); err != nil { + rows.Close() + return stats, err + } + existing[path] = struct{ id, sha string }{id, sha} + } + rows.Close() + + // Remove documents whose files disappeared (chunks cascade). + for path, ex := range existing { + if _, ok := seen[path]; ok { + continue + } + if _, err := s.st.Pool.Exec(ctx, `DELETE FROM cc_documents WHERE id = $1`, ex.id); err != nil { + return stats, fmt.Errorf("delete vanished doc %s: %w", path, err) + } + stats.Removed++ + } + + for _, d := range seen { + if ex, ok := existing[d.rel]; ok && ex.sha == d.sha { + stats.Unchanged++ + continue + } + raw, err := os.ReadFile(d.full) + if err != nil { + return stats, fmt.Errorf("read %s: %w", d.rel, err) + } + failed, n, err := s.syncOne(ctx, d.rel, d.sha, raw) + if err != nil { + return stats, err + } + stats.EmbedFailed = stats.EmbedFailed || failed + stats.Chunks += n + if _, ok := existing[d.rel]; ok { + stats.Updated++ + } else { + stats.Added++ + } + } + return stats, nil +} + +// syncOne upserts one document row and re-chunks it. Returns +// (embedFailed, newChunkCount, err). +func (s *Syncer) syncOne(ctx context.Context, rel, sha string, raw []byte) (bool, int, error) { + fm, body := splitFrontMatter(string(raw)) + title := docTitle(fm, body, rel) + fmJSON, _ := json.Marshal(fm) + + var docID string + err := s.st.Pool.QueryRow(ctx, ` + INSERT INTO cc_documents (path, title, taxonomy, frontmatter, sha256, synced_at) + VALUES ($1,$2,$3,$4,$5,now()) + ON CONFLICT (path) DO UPDATE SET + title = EXCLUDED.title, taxonomy = EXCLUDED.taxonomy, + frontmatter = EXCLUDED.frontmatter, sha256 = EXCLUDED.sha256, + synced_at = now() + RETURNING id`, + rel, title, taxonomyOf(rel), fmJSON, sha).Scan(&docID) + if err != nil { + return false, 0, fmt.Errorf("upsert doc %s: %w", rel, err) + } + failed, n, err := s.rechunk(ctx, rel, body, "doc", docID, "") + return failed, n, err +} + +// IndexArtifact chunks and embeds one cc_stage_artifacts row +// (doc_id NULL, chunk_hash prefixed "artifact:"). +func (s *Syncer) IndexArtifact(ctx context.Context, artifactID string) error { + var content string + err := s.st.Pool.QueryRow(ctx, + `SELECT content_md FROM cc_stage_artifacts WHERE id = $1`, artifactID).Scan(&content) + if err != nil { + return fmt.Errorf("load artifact %s: %w", artifactID, err) + } + failed, _, err := s.rechunk(ctx, artifactID, content, "artifact", "", artifactID) + if failed { + slog.Warn("artifact indexed without embeddings", "artifact", artifactID) + } + return err +} + +// rechunk diffs chunk_hash sets for an owner (doc or artifact) and inserts +// only new chunks; unchanged chunks keep their embeddings. +// ownerKind is "doc" or "artifact"; exactly one of docID/artifactID is set. +func (s *Syncer) rechunk(ctx context.Context, sourceRef, content, ownerKind, docID, artifactID string) (bool, int, error) { + chunks := chunkMarkdown(content) + + existing := map[string]bool{} + q := `SELECT chunk_hash FROM cc_doc_chunks WHERE doc_id = $1` + arg := any(docID) + if ownerKind == "artifact" { + q = `SELECT chunk_hash FROM cc_doc_chunks WHERE artifact_id = $1` + arg = artifactID + } + rows, err := s.st.Pool.Query(ctx, q, arg) + if err != nil { + return false, 0, fmt.Errorf("list chunks: %w", err) + } + for rows.Next() { + var h string + if err := rows.Scan(&h); err != nil { + rows.Close() + return false, 0, err + } + existing[h] = true + } + rows.Close() + + want := make(map[string]Chunk, len(chunks)) + var fresh []Chunk + for _, c := range chunks { + h := chunkHash(ownerKind, sourceRef, c.Seq, c.Content) + want[h] = c + if !existing[h] { + fresh = append(fresh, c) + } + } + + // Drop chunks that are no longer part of the source. + for h := range existing { + if _, ok := want[h]; !ok { + if _, err := s.st.Pool.Exec(ctx, `DELETE FROM cc_doc_chunks WHERE chunk_hash = $1`, h); err != nil { + return false, 0, fmt.Errorf("delete stale chunk: %w", err) + } + } + } + if len(fresh) == 0 { + return false, 0, nil + } + + texts := make([]string, len(fresh)) + for i, c := range fresh { + texts[i] = c.Content + } + var vecs [][]float32 + embedFailed := false + if s.emb != nil { + vecs, err = s.emb.Embed(ctx, texts) + if err != nil { + slog.Warn("embedding unavailable; chunks stored without vectors", + "owner", ownerKind, "ref", sourceRef, "chunks", len(fresh), "err", err) + embedFailed = true + vecs = nil + } + } else { + embedFailed = true + } + + inserted := 0 + for i, c := range fresh { + h := chunkHash(ownerKind, sourceRef, c.Seq, c.Content) + var vecLit any + if !embedFailed && i < len(vecs) { + vecLit = embed.VectorLiteral(vecs[i]) + } + var dID, aID any + if ownerKind == "doc" { + dID = docID + } else { + aID = artifactID + } + _, err := s.st.Pool.Exec(ctx, ` + INSERT INTO cc_doc_chunks (doc_id, artifact_id, seq, heading, content, chunk_hash, embedding) + VALUES ($1,$2,$3,$4,$5,$6, $7::vector) + ON CONFLICT (chunk_hash) DO NOTHING`, + dID, aID, c.Seq, c.Heading, c.Content, h, vecLit) + if err != nil { + return embedFailed, inserted, fmt.Errorf("insert chunk %d: %w", c.Seq, err) + } + inserted++ + } + return embedFailed, inserted, nil +} + +// chunkHash is stable per (owner, source, seq, content); "artifact:" prefix +// keeps artifact chunks distinct from document chunks. +func chunkHash(ownerKind, sourceRef string, seq int, content string) string { + prefix := "" + if ownerKind == "artifact" { + prefix = "artifact:" + } + return sha256hex([]byte(fmt.Sprintf("%s%s:%d:%s", prefix, sourceRef, seq, content))) +} + +// StartPeriodic runs Sync immediately, then every interval until ctx is +// cancelled (same ticker pattern as the lease reaper). +func (s *Syncer) StartPeriodic(ctx context.Context, interval time.Duration) { + go func() { + s.runOnce(ctx) + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + s.runOnce(ctx) + } + } + }() +} + +func (s *Syncer) runOnce(ctx context.Context) { + stats, err := s.Sync(ctx) + if err != nil { + slog.Error("doc sync", "err", err) + return + } + slog.Info("doc sync done", + "added", stats.Added, "updated", stats.Updated, "removed", stats.Removed, + "unchanged", stats.Unchanged, "chunks", stats.Chunks, "embed_failed", stats.EmbedFailed) +} + +// splitFrontMatter extracts a leading --- yaml --- block; returns the parsed +// map (possibly empty) and the remaining body. +func splitFrontMatter(md string) (map[string]any, string) { + fm := map[string]any{} + body := md + if rest, ok := strings.CutPrefix(md, "---\n"); ok { + if end := strings.Index(rest, "\n---"); end >= 0 { + raw := rest[:end] + after := rest[end+len("\n---"):] + if after == "" || after[0] == '\n' || after[0] == '\r' { + if err := yaml.Unmarshal([]byte(raw), &fm); err != nil { + slog.Warn("doc sync: bad front matter", "err", err) + fm = map[string]any{} + } else { + body = strings.TrimLeft(after, "\r\n") + } + } + } + } + if fm == nil { + fm = map[string]any{} + } + return fm, body +} + +// docTitle prefers front matter title, then the first level-1 heading, then +// the file name. +func docTitle(fm map[string]any, body, rel string) string { + if t, ok := fm["title"].(string); ok && strings.TrimSpace(t) != "" { + return strings.TrimSpace(t) + } + for _, line := range strings.Split(body, "\n") { + if h, ok := strings.CutPrefix(strings.TrimSpace(line), "# "); ok { + if t := strings.TrimSpace(h); t != "" { + return t + } + } + } + base := filepath.Base(rel) + return strings.TrimSuffix(base, filepath.Ext(base)) +} + +// taxonomyOf returns the first directory segment under .doc/ ("" for files +// directly at the root). +func taxonomyOf(rel string) string { + rest, ok := strings.CutPrefix(rel, ".doc/") + if !ok { + return "" + } + if i := strings.IndexByte(rest, '/'); i > 0 { + return rest[:i] + } + return "" +} + +func sha256hex(b []byte) string { + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} diff --git a/internal/controlcenter/embed/embed.go b/internal/controlcenter/embed/embed.go new file mode 100644 index 00000000..5e38446e --- /dev/null +++ b/internal/controlcenter/embed/embed.go @@ -0,0 +1,195 @@ +// Package embed provides an OpenAI-compatible embeddings client for the +// Control Center document vector store, plus a deterministic stub for tests. +// Self-contained: modeled on internal/kb/embed.go but does not import it. +package embed + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "time" + "unicode/utf8" +) + +const ( + // Dim is the bge-m3 embedding dimension used by cc_doc_chunks.embedding. + Dim = 1024 + + maxBatchItems = 16 + httpTimeout = 30 * time.Second +) + +// Embedder generates embedding vectors for text batches. +type Embedder interface { + Embed(ctx context.Context, texts []string) ([][]float32, error) + Dim() int +} + +// HTTPClient talks to an OpenAI-compatible /embeddings endpoint. +type HTTPClient struct { + endpoint string + apiKey string + model string + dim int + http *http.Client +} + +// NewHTTP builds a client; endpoint may already include /v1 (or the full +// /embeddings suffix). An empty endpoint yields a client whose Embed always +// fails, so callers can treat it as "embedding unavailable". +func NewHTTP(endpoint, apiKey, model string, dim int) *HTTPClient { + if dim <= 0 { + dim = Dim + } + return &HTTPClient{ + endpoint: normalizeEndpoint(endpoint), + apiKey: apiKey, + model: model, + dim: dim, + http: &http.Client{Timeout: httpTimeout}, + } +} + +func (c *HTTPClient) Dim() int { return c.dim } + +// Embed embeds texts in batches of at most maxBatchItems. +func (c *HTTPClient) Embed(ctx context.Context, texts []string) ([][]float32, error) { + if c.endpoint == "" { + return nil, fmt.Errorf("embed: endpoint not configured") + } + if len(texts) == 0 { + return nil, nil + } + out := make([][]float32, len(texts)) + for start := 0; start < len(texts); start += maxBatchItems { + end := start + maxBatchItems + if end > len(texts) { + end = len(texts) + } + vecs, err := c.embedBatch(ctx, texts[start:end]) + if err != nil { + return nil, err + } + copy(out[start:end], vecs) + } + return out, nil +} + +// embedBatch posts one batch; 5xx and network errors are retried once, 4xx +// is returned immediately. +func (c *HTTPClient) embedBatch(ctx context.Context, texts []string) ([][]float32, error) { + var lastErr error + for attempt := 0; attempt < 2; attempt++ { + vecs, retryable, err := c.post(ctx, texts) + if err == nil { + return vecs, nil + } + if !retryable { + return nil, err + } + lastErr = err + } + return nil, lastErr +} + +func (c *HTTPClient) post(ctx context.Context, texts []string) ([][]float32, bool, error) { + body, _ := json.Marshal(map[string]any{"model": c.model, "input": texts}) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.endpoint, bytes.NewReader(body)) + if err != nil { + return nil, false, err + } + req.Header.Set("Content-Type", "application/json") + if c.apiKey != "" { + req.Header.Set("Authorization", "Bearer "+c.apiKey) + } + resp, err := c.http.Do(req) + if err != nil { + return nil, true, fmt.Errorf("embed request: %w", err) + } + defer resp.Body.Close() + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + if resp.StatusCode >= 300 { + retryable := resp.StatusCode >= 500 + return nil, retryable, fmt.Errorf("embed http %d: %s", resp.StatusCode, truncateUTF8(strings.TrimSpace(string(raw)), 300)) + } + var parsed struct { + Data []struct { + Embedding []float32 `json:"embedding"` + Index int `json:"index"` + } `json:"data"` + } + if err := json.Unmarshal(raw, &parsed); err != nil { + return nil, false, fmt.Errorf("embed decode: %w", err) + } + out := make([][]float32, len(texts)) + for _, d := range parsed.Data { + if d.Index >= 0 && d.Index < len(out) { + out[d.Index] = padOrTrim(d.Embedding, c.dim) + } + } + for i := range out { + if out[i] == nil && i < len(parsed.Data) { + out[i] = padOrTrim(parsed.Data[i].Embedding, c.dim) + } + if out[i] == nil { + return nil, false, fmt.Errorf("embed: missing embedding %d", i) + } + } + return out, false, nil +} + +func normalizeEndpoint(endpoint string) string { + endpoint = strings.TrimRight(strings.TrimSpace(endpoint), "/") + if endpoint == "" { + return "" + } + if strings.HasSuffix(endpoint, "/embeddings") { + return endpoint + } + if strings.HasSuffix(endpoint, "/v1") || strings.Contains(endpoint, "/v1/") { + return endpoint + "/embeddings" + } + return endpoint + "/v1/embeddings" +} + +func padOrTrim(v []float32, n int) []float32 { + if len(v) == n { + return v + } + out := make([]float32, n) + copy(out, v) + return out +} + +func truncateUTF8(s string, n int) string { + if n <= 0 || len(s) <= n { + return s + } + for n > 0 && !utf8.RuneStart(s[n]) { + n-- + } + return s[:n] +} + +// VectorLiteral serializes a vector into the pgvector text format +// "[0.1,0.2,...]"; bind it as a string and cast with ::vector. +func VectorLiteral(vec []float32) string { + if len(vec) == 0 { + return "" + } + var sb strings.Builder + sb.WriteByte('[') + for i, f := range vec { + if i > 0 { + sb.WriteByte(',') + } + sb.WriteString(strconv.FormatFloat(float64(f), 'g', -1, 32)) + } + sb.WriteByte(']') + return sb.String() +} diff --git a/internal/controlcenter/embed/embed_test.go b/internal/controlcenter/embed/embed_test.go new file mode 100644 index 00000000..7a4f6d30 --- /dev/null +++ b/internal/controlcenter/embed/embed_test.go @@ -0,0 +1,119 @@ +package embed + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNormalizeEndpoint(t *testing.T) { + cases := map[string]string{ + "http://x:8080": "http://x:8080/v1/embeddings", + "http://x:8080/": "http://x:8080/v1/embeddings", + "http://x:8080/v1": "http://x:8080/v1/embeddings", + "http://x:8080/v1/embeddings": "http://x:8080/v1/embeddings", + "": "", + } + for in, want := range cases { + assert.Equal(t, want, normalizeEndpoint(in), in) + } +} + +func fakeServer(t *testing.T, dim int, status int, calls *atomic.Int32) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if calls != nil { + calls.Add(1) + } + assert.Equal(t, "/v1/embeddings", r.URL.Path) + assert.Equal(t, "Bearer test-key", r.Header.Get("Authorization")) + if status != 200 { + w.WriteHeader(status) + return + } + var req struct { + Model string `json:"model"` + Input []string `json:"input"` + } + require.NoError(t, json.NewDecoder(r.Body).Decode(&req)) + assert.Equal(t, "bge-m3", req.Model) + type item struct { + Embedding []float32 `json:"embedding"` + Index int `json:"index"` + } + data := make([]item, len(req.Input)) + for i := range req.Input { + data[i] = item{Embedding: make([]float32, dim), Index: i} + } + _ = json.NewEncoder(w).Encode(map[string]any{"data": data}) + })) +} + +func TestEmbedBatches(t *testing.T) { + var calls atomic.Int32 + srv := fakeServer(t, 4, 200, &calls) + defer srv.Close() + + c := NewHTTP(srv.URL, "test-key", "bge-m3", 4) + texts := make([]string, maxBatchItems+3) + vecs, err := c.Embed(context.Background(), texts) + require.NoError(t, err) + assert.Len(t, vecs, len(texts)) + for _, v := range vecs { + assert.Len(t, v, 4) + } + assert.Equal(t, int32(2), calls.Load(), "17 texts must be split into 16+1") +} + +func TestEmbedRetriesOn5xxOnce(t *testing.T) { + var calls atomic.Int32 + srv := fakeServer(t, 4, 500, &calls) + defer srv.Close() + + c := NewHTTP(srv.URL, "test-key", "bge-m3", 4) + _, err := c.Embed(context.Background(), []string{"x"}) + require.Error(t, err) + assert.Equal(t, int32(2), calls.Load(), "5xx retried exactly once") +} + +func TestEmbedNoRetryOn4xx(t *testing.T) { + var calls atomic.Int32 + srv := fakeServer(t, 4, 400, &calls) + defer srv.Close() + + c := NewHTTP(srv.URL, "test-key", "bge-m3", 4) + _, err := c.Embed(context.Background(), []string{"x"}) + require.Error(t, err) + assert.Equal(t, int32(1), calls.Load(), "4xx must not be retried") +} + +func TestEmbedEmptyEndpoint(t *testing.T) { + c := NewHTTP("", "", "bge-m3", 4) + _, err := c.Embed(context.Background(), []string{"x"}) + require.Error(t, err) +} + +func TestStubDeterministic(t *testing.T) { + s := NewStub(8) + a, err := s.Embed(context.Background(), []string{"hello", "world"}) + require.NoError(t, err) + b, err := s.Embed(context.Background(), []string{"hello"}) + require.NoError(t, err) + assert.Equal(t, a[0], b[0], "same text, same vector") + assert.NotEqual(t, a[0], a[1], "different texts differ") + assert.Equal(t, 8, s.Dim()) +} + +func TestVectorLiteral(t *testing.T) { + assert.Equal(t, "", VectorLiteral(nil)) + lit := VectorLiteral([]float32{0.5, -1, 1.25}) + assert.True(t, strings.HasPrefix(lit, "[") && strings.HasSuffix(lit, "]")) + assert.Equal(t, "[0.5,-1,1.25]", lit) +} diff --git a/internal/controlcenter/embed/stub.go b/internal/controlcenter/embed/stub.go new file mode 100644 index 00000000..e13331fa --- /dev/null +++ b/internal/controlcenter/embed/stub.go @@ -0,0 +1,58 @@ +package embed + +import ( + "context" + "crypto/sha256" + "encoding/binary" + "math" +) + +// Stub is a deterministic offline Embedder: each text maps to a fixed +// pseudo-random unit-ish vector derived from its sha256, so tests need no +// network and results are stable across runs. +type Stub struct { + dim int +} + +func NewStub(dim int) *Stub { + if dim <= 0 { + dim = Dim + } + return &Stub{dim: dim} +} + +func (s *Stub) Dim() int { return s.dim } + +func (s *Stub) Embed(_ context.Context, texts []string) ([][]float32, error) { + out := make([][]float32, len(texts)) + for i, t := range texts { + out[i] = s.vector(t) + } + return out, nil +} + +func (s *Stub) vector(text string) []float32 { + seed := sha256.Sum256([]byte(text)) + state := binary.LittleEndian.Uint64(seed[:8]) | 1 + vec := make([]float32, s.dim) + for i := range vec { + // xorshift64* + state ^= state >> 12 + state ^= state << 25 + state ^= state >> 27 + v := (state * 2685821657736338717) >> 11 + vec[i] = float32(float64(v)/float64(uint64(1)<<53))*2 - 1 + } + // normalize so cosine distances behave sanely + var norm float64 + for _, f := range vec { + norm += float64(f * f) + } + norm = math.Sqrt(norm) + if norm > 0 { + for i := range vec { + vec[i] /= float32(norm) + } + } + return vec +} diff --git a/internal/controlcenter/server/docs_handlers.go b/internal/controlcenter/server/docs_handlers.go new file mode 100644 index 00000000..dafef5bb --- /dev/null +++ b/internal/controlcenter/server/docs_handlers.go @@ -0,0 +1,307 @@ +package server + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "strconv" + "strings" + "time" + + "chat-agent-system/internal/controlcenter/embed" +) + +const ( + searchDefaultLimit = 20 + searchMaxLimit = 100 + snippetRunes = 300 +) + +// docListItem is one row of GET /api/v1/docs. +type docListItem struct { + ID string `json:"id"` + Path string `json:"path"` + Title string `json:"title"` + Taxonomy string `json:"taxonomy"` + Frontmatter any `json:"frontmatter,omitempty"` + SyncedAt time.Time `json:"synced_at"` + Chunks int `json:"chunks"` +} + +func (s *Server) handleDocList(w http.ResponseWriter, r *http.Request) { + taxonomy := strings.TrimSpace(r.URL.Query().Get("taxonomy")) + q := strings.TrimSpace(r.URL.Query().Get("q")) + like := "%" + escapeLike(q) + "%" + + rows, err := s.store.Pool.Query(r.Context(), ` + SELECT d.id, d.path, d.title, d.taxonomy, d.synced_at, + (SELECT count(*) FROM cc_doc_chunks c WHERE c.doc_id = d.id) AS chunks + FROM cc_documents d + WHERE ($1 = '' OR d.taxonomy = $1) + AND ($2 = '' OR d.path ILIKE $3 ESCAPE '\' OR d.title ILIKE $3 ESCAPE '\') + ORDER BY d.path`, taxonomy, q, like) + if err != nil { + slog.Error("doc list", "err", err) + writeError(w, http.StatusInternalServerError, "internal error") + return + } + defer rows.Close() + items := []docListItem{} + for rows.Next() { + var it docListItem + if err := rows.Scan(&it.ID, &it.Path, &it.Title, &it.Taxonomy, &it.SyncedAt, &it.Chunks); err != nil { + slog.Error("doc list scan", "err", err) + writeError(w, http.StatusInternalServerError, "internal error") + return + } + items = append(items, it) + } + writeJSON(w, http.StatusOK, map[string]any{"items": items, "count": len(items)}) +} + +func (s *Server) handleDocGet(w http.ResponseWriter, r *http.Request) { + var it docListItem + var fm []byte + err := s.store.Pool.QueryRow(r.Context(), ` + SELECT d.id, d.path, d.title, d.taxonomy, d.frontmatter, d.synced_at, + (SELECT count(*) FROM cc_doc_chunks c WHERE c.doc_id = d.id) AS chunks + FROM cc_documents d WHERE d.id = $1`, r.PathValue("id")). + Scan(&it.ID, &it.Path, &it.Title, &it.Taxonomy, &fm, &it.SyncedAt, &it.Chunks) + if err != nil { + writeError(w, http.StatusNotFound, "document not found") + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "id": it.ID, "path": it.Path, "title": it.Title, "taxonomy": it.Taxonomy, + "frontmatter": jsonRaw(fm), "synced_at": it.SyncedAt, "chunks": it.Chunks, + }) +} + +// searchHit is one scored chunk with its source resolved. +type searchHit struct { + chunkID int64 + docID *string + artID *string + heading string + content string + score float64 + docPath string + docTitle string + taskSlug string + artStage string + artTitle string +} + +const searchSelect = ` + SELECT c.id, c.doc_id, c.artifact_id, c.heading, c.content, + COALESCE(d.path,''), COALESCE(d.title,''), + COALESCE(t.slug,''), COALESCE(a.stage,''), COALESCE(a.title,'')` + +const searchJoin = ` + FROM cc_doc_chunks c + LEFT JOIN cc_documents d ON d.id = c.doc_id + LEFT JOIN cc_stage_artifacts a ON a.id = c.artifact_id + LEFT JOIN cc_tasks t ON t.id = a.task_id` + +func (s *Server) semanticHits(ctx context.Context, qvec []float32, limit int) ([]searchHit, error) { + lit := embed.VectorLiteral(qvec) + rows, err := s.store.Pool.Query(ctx, searchSelect+`, + 1 - (c.embedding <=> $1::vector) AS score`+searchJoin+` + WHERE c.embedding IS NOT NULL + ORDER BY c.embedding <=> $1::vector + LIMIT $2`, lit, limit) + if err != nil { + return nil, err + } + defer rows.Close() + return scanHits(rows) +} + +func (s *Server) keywordHits(ctx context.Context, q string, limit int) ([]searchHit, error) { + like := "%" + escapeLike(q) + "%" + rows, err := s.store.Pool.Query(ctx, searchSelect+`, + (CASE WHEN c.tsv @@ plainto_tsquery('simple', $1) + THEN ts_rank(c.tsv, plainto_tsquery('simple', $1)) ELSE 0 END + + CASE WHEN c.content ILIKE $2 ESCAPE '\' THEN 0.5 ELSE 0 END + + CASE WHEN c.heading ILIKE $2 ESCAPE '\' THEN 0.3 ELSE 0 END) AS score`+searchJoin+` + WHERE c.tsv @@ plainto_tsquery('simple', $1) + OR c.content ILIKE $2 ESCAPE '\' + OR c.heading ILIKE $2 ESCAPE '\' + ORDER BY score DESC + LIMIT $3`, q, like, limit) + if err != nil { + return nil, err + } + defer rows.Close() + return scanHits(rows) +} + +func scanHits(rows interface { + Next() bool + Scan(...any) error + Err() error +}) ([]searchHit, error) { + var out []searchHit + for rows.Next() { + var h searchHit + if err := rows.Scan(&h.chunkID, &h.docID, &h.artID, &h.heading, &h.content, + &h.docPath, &h.docTitle, &h.taskSlug, &h.artStage, &h.artTitle, &h.score); err != nil { + return nil, err + } + out = append(out, h) + } + return out, rows.Err() +} + +func (s *Server) handleDocSearch(w http.ResponseWriter, r *http.Request) { + q := strings.TrimSpace(r.URL.Query().Get("q")) + if q == "" { + writeError(w, http.StatusBadRequest, "q is required") + return + } + mode := r.URL.Query().Get("mode") + if mode == "" { + mode = "hybrid" + } + if mode != "semantic" && mode != "keyword" && mode != "hybrid" { + writeError(w, http.StatusBadRequest, "mode must be semantic|keyword|hybrid") + return + } + limit := searchDefaultLimit + if raw := r.URL.Query().Get("limit"); raw != "" { + n, err := strconv.Atoi(raw) + if err != nil || n < 1 { + writeError(w, http.StatusBadRequest, "limit must be a positive integer") + return + } + limit = min(n, searchMaxLimit) + } + + ctx := r.Context() + degraded := false + var vecHits []searchHit + if mode != "keyword" { + if s.emb == nil { + degraded = true + } else if vecs, err := s.emb.Embed(ctx, []string{q}); err != nil || len(vecs) == 0 { + slog.Warn("search: query embedding failed, degrading to keyword", "err", err) + degraded = true + } else if hits, err := s.semanticHits(ctx, vecs[0], limit); err != nil { + slog.Error("search: semantic query", "err", err) + writeError(w, http.StatusInternalServerError, "internal error") + return + } else { + vecHits = hits + } + } + + var hits []searchHit + switch { + case mode == "semantic" && !degraded: + hits = vecHits + case mode == "hybrid" && !degraded: + kwHits, err := s.keywordHits(ctx, q, limit) + if err != nil { + slog.Error("search: keyword query", "err", err) + writeError(w, http.StatusInternalServerError, "internal error") + return + } + hits = fuseHits(vecHits, kwHits, limit) + default: // keyword, or degraded semantic/hybrid + kwHits, err := s.keywordHits(ctx, q, limit) + if err != nil { + slog.Error("search: keyword query", "err", err) + writeError(w, http.StatusInternalServerError, "internal error") + return + } + hits = kwHits + } + + results := make([]map[string]any, 0, len(hits)) + for _, h := range hits { + source := map[string]any{"type": "doc", "path": h.docPath, "title": h.docTitle} + if h.artID != nil { + source = map[string]any{"type": "artifact", "task_slug": h.taskSlug, "stage": h.artStage, "title": h.artTitle} + } + results = append(results, map[string]any{ + "score": h.score, + "snippet": snippet(h.content), + "heading": h.heading, + "source": source, + "doc_id": h.docID, + "artifact_id": h.artID, + }) + } + writeJSON(w, http.StatusOK, map[string]any{ + "q": q, "mode": mode, "degraded": degraded, "count": len(results), "results": results, + }) +} + +// fuseHits merges semantic (weight 0.6) and keyword (weight 0.4) hits after +// per-list max-normalization. +func fuseHits(vecHits, kwHits []searchHit, limit int) []searchHit { + byID := map[int64]*searchHit{} + norm := func(hits []searchHit, weight float64) { + maxScore := 0.0 + for _, h := range hits { + if h.score > maxScore { + maxScore = h.score + } + } + if maxScore <= 0 { + maxScore = 1 + } + for _, h := range hits { + score := weight * h.score / maxScore + if ex, ok := byID[h.chunkID]; ok { + ex.score += score + } else { + cpy := h + cpy.score = score + byID[h.chunkID] = &cpy + } + } + } + norm(vecHits, 0.6) + norm(kwHits, 0.4) + + out := make([]searchHit, 0, len(byID)) + for _, h := range byID { + out = append(out, *h) + } + for i := 1; i < len(out); i++ { // insertion sort by score desc, hits are few + for j := i; j > 0 && out[j].score > out[j-1].score; j-- { + out[j], out[j-1] = out[j-1], out[j] + } + } + if len(out) > limit { + out = out[:limit] + } + return out +} + +// snippet returns the first ~snippetRunes runes of content. +func snippet(content string) string { + r := []rune(strings.TrimSpace(content)) + if len(r) > snippetRunes { + r = r[:snippetRunes] + } + return string(r) +} + +func escapeLike(s string) string { + r := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`) + return r.Replace(s) +} + +// jsonRaw returns frontmatter bytes as an any for the JSON encoder. +func jsonRaw(b []byte) any { + var v any + if len(b) > 0 { + if err := json.Unmarshal(b, &v); err == nil { + return v + } + } + return map[string]any{} +} diff --git a/internal/controlcenter/server/docs_test.go b/internal/controlcenter/server/docs_test.go new file mode 100644 index 00000000..fd7dabd8 --- /dev/null +++ b/internal/controlcenter/server/docs_test.go @@ -0,0 +1,213 @@ +package server + +import ( + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "chat-agent-system/internal/controlcenter/config" + "chat-agent-system/internal/controlcenter/docs" + "chat-agent-system/internal/controlcenter/embed" + "chat-agent-system/internal/controlcenter/store" +) + +// newDocTestServer builds a server wired to a docs.Syncer over a temp .doc +// tree. emb may be nil to exercise degraded search. +func newDocTestServer(t *testing.T, emb embed.Embedder, files map[string]string) (*httptest.Server, *store.Store, *docs.Syncer) { + t.Helper() + dsn := os.Getenv("TEST_CC_DB_DSN") + if dsn == "" { + t.Skip("TEST_CC_DB_DSN not set") + } + _, thisFile, _, _ := runtime.Caller(0) + migrationsDir := filepath.Join(filepath.Dir(thisFile), "..", "..", "..", "deploy", "control-center", "goose") + ctx := t.Context() + require.NoError(t, store.Migrate(ctx, dsn, migrationsDir)) + st, err := store.Connect(ctx, dsn) + require.NoError(t, err) + t.Cleanup(st.Close) + + repoRoot := t.TempDir() + for rel, content := range files { + full := filepath.Join(repoRoot, filepath.FromSlash(rel)) + require.NoError(t, os.MkdirAll(filepath.Dir(full), 0o755)) + require.NoError(t, os.WriteFile(full, []byte(content), 0o644)) + } + + syncer := docs.New(st, repoRoot, emb) + stats, err := syncer.Sync(ctx) + require.NoError(t, err) + + srv := New(config.Config{ConsoleToken: testConsoleToken}, st) + srv.SetDocs(emb, syncer) + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(ts.Close) + t.Logf("sync stats: %+v", stats) + return ts, st, syncer +} + +func docTestFiles(tag string) map[string]string { + return map[string]string{ + ".doc/ops/" + tag + "-pipeline.md": "---\ntitle: 流水线设计\nstatus: done\n---\n# 流水线设计\n\n八阶段流水线,任务从 proposal 到 confirmation。\n\n## 阶段\n\n提案、分析、设计、评审、实现、测试、评估、确认。\n", + ".doc/audit/" + tag + "-notes.md": "# 审计笔记\n\n流水线租约过期后任务回到 assigned 状态。\n", + ".doc/" + tag + "-root.md": "# 根文档\n\n没有 taxonomy 的根级文档。\n", + ".doc/assets/" + tag + "-skip.md": "# 应被跳过\n\nassets 目录不入库。\n", + ".doc/ops/" + tag + "-plain.txt": "not markdown, ignored", + } +} + +func TestDocsSyncListAndGet(t *testing.T) { + tag := fmt.Sprintf("t%d", time.Now().UnixNano()) + ts, st, syncer := newDocTestServer(t, embed.NewStub(embed.Dim), docTestFiles(tag)) + + code, raw := doReq(t, "GET", ts.URL+"/api/v1/docs", testConsoleToken, nil) + require.Equal(t, http.StatusOK, code, string(raw)) + items := decodeBody(t, raw)["items"].([]any) + var paths []string + for _, it := range items { + paths = append(paths, it.(map[string]any)["path"].(string)) + } + assert.Contains(t, paths, ".doc/ops/"+tag+"-pipeline.md") + assert.Contains(t, paths, ".doc/audit/"+tag+"-notes.md") + assert.Contains(t, paths, ".doc/"+tag+"-root.md") + assert.NotContains(t, paths, ".doc/assets/"+tag+"-skip.md", "assets/ must be skipped") + + // taxonomy filter + code, raw = doReq(t, "GET", ts.URL+"/api/v1/docs?taxonomy=audit", testConsoleToken, nil) + require.Equal(t, http.StatusOK, code) + for _, it := range decodeBody(t, raw)["items"].([]any) { + assert.Equal(t, "audit", it.(map[string]any)["taxonomy"]) + } + + // q filter + front matter title + code, raw = doReq(t, "GET", ts.URL+"/api/v1/docs?q=流水线设计", testConsoleToken, nil) + require.Equal(t, http.StatusOK, code) + items = decodeBody(t, raw)["items"].([]any) + require.Len(t, items, 1) + it := items[0].(map[string]any) + assert.Equal(t, "流水线设计", it["title"]) + assert.Equal(t, "ops", it["taxonomy"]) + assert.Greater(t, it["chunks"].(float64), float64(0)) + + // detail + code, raw = doReq(t, "GET", ts.URL+"/api/v1/docs/"+it["id"].(string), testConsoleToken, nil) + require.Equal(t, http.StatusOK, code, string(raw)) + detail := decodeBody(t, raw) + assert.Equal(t, "done", detail["frontmatter"].(map[string]any)["status"]) + + code, _ = doReq(t, "GET", ts.URL+"/api/v1/docs/00000000-0000-0000-0000-000000000000", testConsoleToken, nil) + assert.Equal(t, http.StatusNotFound, code) + + // re-sync is idempotent: chunk count unchanged + countChunks := func() int { + var n int + require.NoError(t, st.Pool.QueryRow(t.Context(), + `SELECT count(*) FROM cc_doc_chunks c JOIN cc_documents d ON d.id = c.doc_id + WHERE d.path LIKE '%' || $1 || '%'`, tag).Scan(&n)) + return n + } + before := countChunks() + stats, err := syncer.Sync(t.Context()) + require.NoError(t, err) + assert.Equal(t, 0, stats.Added+stats.Updated, "unchanged files must be skipped") + assert.Equal(t, before, countChunks(), "unchanged docs must not duplicate chunks") +} + +func TestDocsSearchModes(t *testing.T) { + tag := fmt.Sprintf("t%d", time.Now().UnixNano()) + ts, _, _ := newDocTestServer(t, embed.NewStub(embed.Dim), docTestFiles(tag)) + + // q required + code, _ := doReq(t, "GET", ts.URL+"/api/v1/docs/search", testConsoleToken, nil) + assert.Equal(t, http.StatusBadRequest, code) + + // keyword: ILIKE fallback hits Chinese content + code, raw := doReq(t, "GET", ts.URL+"/api/v1/docs/search?q=流水线&mode=keyword", testConsoleToken, nil) + require.Equal(t, http.StatusOK, code, string(raw)) + m := decodeBody(t, raw) + assert.Equal(t, false, m["degraded"]) + results := m["results"].([]any) + require.NotEmpty(t, results) + top := results[0].(map[string]any) + src := top["source"].(map[string]any) + assert.Equal(t, "doc", src["type"]) + assert.Contains(t, src["path"], tag) + assert.LessOrEqual(t, len([]rune(top["snippet"].(string))), snippetRunes) + + // semantic with stub embedder returns results + code, raw = doReq(t, "GET", ts.URL+"/api/v1/docs/search?q=pipeline&mode=semantic&limit=5", testConsoleToken, nil) + require.Equal(t, http.StatusOK, code, string(raw)) + m = decodeBody(t, raw) + assert.Equal(t, false, m["degraded"]) + assert.NotEmpty(t, m["results"]) + + // hybrid fuses both + code, raw = doReq(t, "GET", ts.URL+"/api/v1/docs/search?q=流水线&mode=hybrid", testConsoleToken, nil) + require.Equal(t, http.StatusOK, code, string(raw)) + m = decodeBody(t, raw) + assert.Equal(t, false, m["degraded"]) + assert.NotEmpty(t, m["results"]) + + // limit cap + code, raw = doReq(t, "GET", ts.URL+"/api/v1/docs/search?q=流水线&limit=500", testConsoleToken, nil) + require.Equal(t, http.StatusOK, code, string(raw)) + assert.LessOrEqual(t, len(decodeBody(t, raw)["results"].([]any)), searchMaxLimit) +} + +func TestDocsSearchDegradedWithoutEmbedder(t *testing.T) { + tag := fmt.Sprintf("t%d", time.Now().UnixNano()) + ts, _, _ := newDocTestServer(t, nil, docTestFiles(tag)) // nil embedder + + code, raw := doReq(t, "GET", ts.URL+"/api/v1/docs/search?q=流水线&mode=hybrid", testConsoleToken, nil) + require.Equal(t, http.StatusOK, code, string(raw)) + m := decodeBody(t, raw) + assert.Equal(t, true, m["degraded"], "no embedder must degrade hybrid to keyword") + assert.NotEmpty(t, m["results"], "keyword fallback still returns hits") +} + +func TestArtifactIndexingHook(t *testing.T) { + tag := fmt.Sprintf("t%d", time.Now().UnixNano()) + dsn := os.Getenv("TEST_CC_DB_DSN") + if dsn == "" { + t.Skip("TEST_CC_DB_DSN not set") + } + ts, st, _ := newDocTestServer(t, embed.NewStub(embed.Dim), docTestFiles(tag)) + require.NoError(t, RegisterAgent(t.Context(), st, "agent-a", "test", testAgentToken)) + + id := createAssignedTask(t, ts, true) + claimTask(t, ts, id) + code, raw := doReq(t, "POST", ts.URL+"/api/v1/agent/tasks/"+id+"/artifacts", testAgentToken, map[string]any{ + "title": "分析产物", "content_md": "# 分析\n\n流水线分析产物 " + tag + " 内容。\n", + }) + require.Equal(t, http.StatusCreated, code, string(raw)) + artID := decodeBody(t, raw)["id"].(string) + + // indexing is async; poll + var n int + for i := 0; i < 40; i++ { + require.NoError(t, st.Pool.QueryRow(t.Context(), + `SELECT count(*) FROM cc_doc_chunks WHERE artifact_id = $1`, artID).Scan(&n)) + if n > 0 { + break + } + time.Sleep(100 * time.Millisecond) + } + assert.Greater(t, n, 0, "artifact chunks indexed asynchronously") + + // artifact shows up in search with its source + code, raw = doReq(t, "GET", ts.URL+"/api/v1/docs/search?q="+tag+"&mode=keyword", testConsoleToken, nil) + require.Equal(t, http.StatusOK, code, string(raw)) + results := decodeBody(t, raw)["results"].([]any) + require.NotEmpty(t, results) + src := results[0].(map[string]any)["source"].(map[string]any) + assert.Equal(t, "artifact", src["type"]) + assert.NotEmpty(t, src["task_slug"]) +} diff --git a/internal/controlcenter/server/server.go b/internal/controlcenter/server/server.go index c13aee5a..d6bf82dc 100644 --- a/internal/controlcenter/server/server.go +++ b/internal/controlcenter/server/server.go @@ -12,8 +12,11 @@ import ( "log/slog" "net/http" "strings" + "time" "chat-agent-system/internal/controlcenter/config" + "chat-agent-system/internal/controlcenter/docs" + "chat-agent-system/internal/controlcenter/embed" "chat-agent-system/internal/controlcenter/store" ) @@ -28,11 +31,13 @@ func AgentFromCtx(ctx context.Context) string { } type Server struct { - cfg config.Config - store *store.Store - agents *AgentAuthenticator - mux *http.ServeMux - broker *Broker + cfg config.Config + store *store.Store + agents *AgentAuthenticator + mux *http.ServeMux + broker *Broker + emb embed.Embedder // nil = search degrades to keyword-only + docSync *docs.Syncer // nil = artifact indexing disabled } func New(cfg config.Config, st *store.Store) *Server { @@ -84,6 +89,31 @@ func (s *Server) routes() { s.mux.HandleFunc("PUT /api/v1/stage-settings", s.handleStageSettingsPut) s.mux.HandleFunc("GET /api/v1/overview", s.handleOverview) s.mux.HandleFunc("GET /api/v1/stream", s.handleStream) + s.mux.HandleFunc("GET /api/v1/docs", s.handleDocList) + s.mux.HandleFunc("GET /api/v1/docs/search", s.handleDocSearch) + s.mux.HandleFunc("GET /api/v1/docs/{id}", s.handleDocGet) +} + +// SetDocs attaches the embedding client and document syncer used by the +// /api/v1/docs search routes and asynchronous artifact indexing. +func (s *Server) SetDocs(emb embed.Embedder, syncer *docs.Syncer) { + s.emb = emb + s.docSync = syncer +} + +// indexArtifactAsync indexes an artifact's markdown in the background; +// failures are logged only. +func (s *Server) indexArtifactAsync(artifactID string) { + if s.docSync == nil { + return + } + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + if err := s.docSync.IndexArtifact(ctx, artifactID); err != nil { + slog.Error("artifact indexing", "artifact", artifactID, "err", err) + } + }() } func (s *Server) Store() *store.Store { return s.store } diff --git a/internal/controlcenter/server/tasks_handlers.go b/internal/controlcenter/server/tasks_handlers.go index 078eab40..1dba13d1 100644 --- a/internal/controlcenter/server/tasks_handlers.go +++ b/internal/controlcenter/server/tasks_handlers.go @@ -164,6 +164,7 @@ func (s *Server) handleAgentArtifact(w http.ResponseWriter, r *http.Request) { writeDomainError(w, err) return } + s.indexArtifactAsync(a.ID) writeJSON(w, http.StatusCreated, map[string]string{"id": a.ID}) } @@ -407,6 +408,7 @@ func (s *Server) handleTaskExternalSubmit(w http.ResponseWriter, r *http.Request writeDomainError(w, err) return } + s.indexArtifactAsync(a.ID) resp := map[string]any{"artifact": a, "advanced": false} if body.Advance { t, done, err := tasks.AdvanceStage(r.Context(), s.store, id, "external", "user-external", body.Note,