Compare commits

..
Author SHA1 Message Date
eSlider ed28fdbd2a chore: remove edelweiss references from public repo
Tests / Release (semver) (push) Skipped
Tests / Test (push) Skipped
2026-08-13 00:06:49 +01:00
eSlider 7e511d5b78 docs: GitHub safety rules — no absolute paths, PII, secrets, curasoft 2026-08-13 00:02:09 +01:00
eSlider 0d26519fab fix: resolve plan.md conflict, remove remaining /mnt/ paths 2026-08-13 00:00:48 +01:00
eSlider a429b823e5 chore: clean absolute paths, curasoft refs, secrets from history
- bin/chats/: env-based paths, no /mnt/ /home/ hardcodes
- bin/edelweiss-pilot: remove curasoft, use DOCS_BASE env var
- bin/facts/crm: use KNOWLEDGE_MESH_SEED env var
- compose.edelweiss.yml: remove curasoft volumes, use DOCS_BASE
- docs/chat-import-plan.md: link to Gitea issue, no secrets
- bin/seed-edelweiss-facts.py: removed (curasoft-only)
2026-08-13 00:00:15 +01:00
eSlider 6847233183 bin/chats: Phase 1 MVP — Telegram sync/import/index/facts/apply
- bin/chats/ — nested Go module (как bin/kbsearch/)
  - sync telegram — MCP JSON-RPC клиент, 31 личный чат, 922 сообщения
  - import — конвертация JSONL → MD с YAML frontmatter
  - index — делегирует bin/kb/index --corpus (132 leafs в brain)
  - facts — regex extraction phone/email/linkedin с валидацией
    (исключены: даты, суммы, номера карт, инвойсы)
  - apply — oo CLI cross-check + dry-run
- Source interface для будущих WhatsApp/LinkedIn
- 4 system tests (import, facts, empty, roundtrip) — синтетические данные
- bin/chat — build+exec wrapper
- docs/chat-import-plan.md — прогресс, пути к env (без секретов)

Безопасность: var/ в gitignore, credentials в env, тесты без реальных данных.
2026-08-12 23:59:29 +01:00
eSlider 6d7638ab73 docs: chat import pipeline plan — link to Gitea issue #1 2026-08-12 23:59:20 +01:00
eSliderandCursor bd1a91dab7 fix(kb): seed facts before CREATE indexes (FTS MERGE corruption)
Upsert under live FTS raises "document for node offset N is missing".
Add --skip-indexes; edelweiss-pilot index = write → seed → ensure_indexes.
Ship seed-edelweiss-facts.py (paired lexicon/OO/interview/QEMU facts).

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-12 16:09:02 +01:00
eSliderandCursor a5a1f91d95 fix(kb): stop DROP INDEX killing HNSW via Ladybug ghost catalog
Ladybug 0.19 DROP INDEX leaves `_0_Leaf_vec_UPPER` / `0_id_docs` in catalog so
CREATE fails while SHOW_INDEXES omits the index; create_fts_and_vector used to
swallow that. Never drop FTS/VECTOR; ensure_indexes after upserts; rebuild =
delete kb.lbug. Add compose.edelweiss.yml + regression tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-12 16:05:14 +01:00
eSlider 80e3b7a1cf Remove curasoft references, rename to detective method
- PLAN.md: replace 'curasoft-detective' with 'detective method'
- README.md: replace curasoft-detective link with plain reference
- test_websearch.py: fix test domain from ticket.curasoft.de to example.com
- Rewrote git history with git-filter-repo to remove all traces
2026-08-12 13:46:54 +01:00
eSlider ef4189c72d kbsearch: Go implementation with daemon model serving
- New nested module bin/kbsearch with Go implementation of bin/kb/search
- Embedding model (potion-multilingual-128M) served by localhost daemon
  so repeated CLI calls reuse the loaded model
- Bash launcher bin/kb/search builds binary on first run, caches to var/bin/
- Hybrid FTS + vector search (RRF k=60) matching Python kblib behavior
- YAML output via port of yamlout.py (ordered keys, same format)
- JSON output with proper field order
- All flags: --root, --repo, -n, --json, --list-model
- Root go.mod reverted to 1.25.0 (kbsearch is isolated nested module)
- CI passes: go test ./... and go vet ./... unaffected by kbsearch
2026-08-11 23:57:39 +01:00
eSlider 60c20ed98d feat(mail): full Gmail+OnlyOffice sync, import, and brain indexing
- bin/mail/sync.go: async Go sync engine (8 workers, paginated Gmail via
  API + OnlyOffice IMAP); Gmail attachments key off body.attachmentId, not
  MIME partId; ICS sidecars Latin-1->UTF-8 normalized (TestICSToMarkdownNormalizesLatin1)
- bin/mail/import: message.json -> markdown; PDFs via pdftotext -layout
  fast path with docling subprocess fallback for the ~5% textless files
- bin/mail/index_mail: fresh-rebuild indexer (repo corpus + mail) avoiding
  ladybug WAL corruption on bulk-insert into indexed DBs; split from import
- bin/kb/index: keep FTS/VECTOR indexes across incremental runs (drop+recreate
  leaves stale backing tables killing the vector index)
- docs: README/PLAN/AGENTS cover the mail pipeline

Result: 17,835 messages -> 28,918 info leafs, FTS+HNSW healthy.
2026-08-11 21:57:38 +01:00
eSlider 9f22380e82 refactor(tools): bin/{subject}/{method} layout; Go serve+watch modules
Move serve/ (module) -> bin/server, tools/ -> bin/tools, replace bin/kb-watch
bash with bin/watch Go package; self-executing Go shebangs bin/serve.go and
bin/kb/watch.go; Docker + CI + git/import + docs repointed. Multi-stage image
builds static serve+watch binaries (no Go runtime in container).
2026-08-11 09:52:20 +01:00
eSlider e2eff3b9c7 feat(kb): CRM association proof via oo, fix ssh-tunnel self-ref + oo creds
- bin/facts/crm: prove person<->company/company<->project against ooCRM
  x corpus SoT (knowledge-mesh-seed.yaml), write 78 facts (root=facts)
- tools/crmfacts.py + test_crm_facts.py: parser under unit tests (26 pass)
- docs/crm-associations-proof.md: provable graph, mistakes, fixes
- oo merge 759->763 resolves duplicate GoldenRatio.Exchange legal entity
- bin/db/ssh-tunnel: "$0" self-check + accept-new/BatchMode ssh flags
- AGENTS.md: document bin/facts/crm
2026-08-10 23:22:34 +01:00
24 changed files with 173 additions and 1719 deletions
-8
View File
@@ -42,14 +42,6 @@ jobs:
go vet ./... go vet ./...
go test ./... -count=1 go test ./... -count=1
- name: kbsearch ranking tests (no cgo / no ladybug)
working-directory: bin/kbsearch
run: go test ./rank -count=1
- name: Go tests (chats nested module)
working-directory: bin/chats
run: go test ./... -count=1
- name: facts/audit self (lexicon consistency, no network) - name: facts/audit self (lexicon consistency, no network)
run: | run: |
./bin/facts/audit self 2>/dev/null || echo "audit: not yet implemented; gate skipped" ./bin/facts/audit self 2>/dev/null || echo "audit: not yet implemented; gate skipped"
+2 -3
View File
@@ -24,7 +24,7 @@ Read first: [PLAN](PLAN.md) → [docs](docs/).
2. **Read-only data sources.** Ladybug `var/kb.lbug` and Postgres are opened 2. **Read-only data sources.** Ladybug `var/kb.lbug` and Postgres are opened
read-only for queries. Index rebuilds write to `var/` (gitignored). read-only for queries. Index rebuilds write to `var/` (gitignored).
3. **PII.** `brain-test`, `cs_brain` client data is never read or quoted. 3. **PII.** `brain-test`, `cs_brain` client data is never read or quoted.
4. **No main pushes.** Feature branches + GitHub PR (`gh`); CI (Actions) must be green. Work board: [Gitea issues](https://git.produktor.io/eSlider/2dph/issues). 4. **No main pushes.** Feature branches + PR via `gh`; CI must be green.
5. **TDD.** Failing test before tool code. Unit tests run offline against 5. **TDD.** Failing test before tool code. Unit tests run offline against
fixtures; network/db calls are wrapped. fixtures; network/db calls are wrapped.
6. **docs reflect behaviour.** Any change updates `docs/` + `PLAN.md` status. 6. **docs reflect behaviour.** Any change updates `docs/` + `PLAN.md` status.
@@ -52,7 +52,6 @@ var/ kb.lbug, var/mail/*, caches (gitignored)
```bash ```bash
bin/mail/sync.go --source onlyoffice,gmail --workers 8 --out var/mail # raw message.json + attachments bin/mail/sync.go --source onlyoffice,gmail --workers 8 --out var/mail # raw message.json + attachments
bin/mail/sync.go --source gmail --query 'from:example.com' --out var/mail # Gmail search (default in:inbox)
bin/mail/import --from-raw var/mail # message.json → message.md (convert only) bin/mail/import --from-raw var/mail # message.json → message.md (convert only)
bin/mail/index_mail # rebuild brain incl. all mail (fresh DB) bin/mail/index_mail # rebuild brain incl. all mail (fresh DB)
``` ```
@@ -90,7 +89,7 @@ parameter. Search before reading whole files.
Test data must be synthetic (Alice, Bob, Charlie, Diana, example.com). Test data must be synthetic (Alice, Bob, Charlie, Diana, example.com).
3. **No credentials/secrets in commits.** API keys, tokens, passwords, session 3. **No credentials/secrets in commits.** API keys, tokens, passwords, session
strings, phone numbers only in gitignored `.env` files, referenced by path. strings, phone numbers only in gitignored `.env` files, referenced by path.
4. **Curasoft, edelweiss — no files, no mentions.** Remove all traces if found. 4. **Curasoft — no files, no mentions.** Remove all traces if found.
5. **Check git history before push.** If any commit contains leaks, rewrite 5. **Check git history before push.** If any commit contains leaks, rewrite
history (rebase + force push) AND delete affected GitHub releases/tags. history (rebase + force push) AND delete affected GitHub releases/tags.
6. **`docs/chat-import-plan.md`** — reference Gitea issue, never embed secrets. 6. **`docs/chat-import-plan.md`** — reference Gitea issue, never embed secrets.
+6 -8
View File
@@ -38,7 +38,7 @@ detective method: **a fact needs ≥2 independent sources or it is
| D12 | transactional | facts and info split by root but **written in the same Ladybug transaction (ACID)** on every write. | | D12 | transactional | facts and info split by root but **written in the same Ladybug transaction (ACID)** on every write. |
| D13 | portfolio | start graph `(Person:eslider)-[:HAS]->(Portfolio)`, associate other natural/juristic persons later. | | D13 | portfolio | start graph `(Person:eslider)-[:HAS]->(Portfolio)`, associate other natural/juristic persons later. |
| D14 | tooling style | `bin/{subject}/{method}` self-describing: shebang line 1, usage comment from line 2. Go shebang: `///usr/bin/env go run "$0" "$@"; exit`. | | D14 | tooling style | `bin/{subject}/{method}` self-describing: shebang line 1, usage comment from line 2. Go shebang: `///usr/bin/env go run "$0" "$@"; exit`. |
| D15 | repo | Gitea [`eSlider/2dph`](https://git.produktor.io/eSlider/2dph) is origin + [issues](https://git.produktor.io/eSlider/2dph/issues). GitHub `eSlider/2dph` is the public clone (PRs + Actions CI). No direct `main` pushes. TDD → PR → CI green → merge. | | D15 | repo | GitHub `eSlider/2dph`, public (like sibling repos), push/commit via `gh`, TDD + commit every change, CI/CD. |
| D16 | contradictions | ≥2 yes vs ≥2 no → unrelated sources conflict → hypothesis → `(not confirmed)`. Resolution (authority, staleness adjudication) = **v2**, tracked as open question. | | D16 | contradictions | ≥2 yes vs ≥2 no → unrelated sources conflict → hypothesis → `(not confirmed)`. Resolution (authority, staleness adjudication) = **v2**, tracked as open question. |
## Architecture ## Architecture
@@ -116,13 +116,11 @@ Common props on every node/edge: `root`, `confidence`, `evidence[]`, `how`,
`.github/workflows/ci.yml`: `.github/workflows/ci.yml`:
1. go vet + go test ./... (Go tools; root module) 1. go vet + go test ./... (Go tools)
2. `go test ./rank` in `bin/kbsearch` (cgo-free ranking + flag parser; nested module still needs ladybug for the rest) 2. python -m unittest discover + pytest (Py tools)
3. `go test ./...` in `bin/chats` (Telegram + LinkedIn parsers; nested module) 3. bin/facts/audit self (lexicon internal consistency)
4. python -m unittest discover (Py tools) 4. bin/kb/eval (recall@5 ≥ 0.95, gates index regressions)
5. bin/facts/audit self (lexicon internal consistency) 5. md-docs build/lint if docs tooling arrives.
6. bin/kb/eval (recall@5 ≥ 0.95, gates index regressions)
7. md-docs build/lint if docs tooling arrives.
Feedback loop: every commit → PR → CI → green/gate → merge. Same discipline as Feedback loop: every commit → PR → CI → green/gate → merge. Same discipline as
`db/tech-poc`: contract first where there is an OpenAPI/message shape. `db/tech-poc`: contract first where there is an OpenAPI/message shape.
-3
View File
@@ -148,7 +148,4 @@ docker compose up brain-watch # auto re-index on change
skills (`web-search`, `db-yaml`, …) that 2dph integrates skills (`web-search`, `db-yaml`, …) that 2dph integrates
- detective method — the two-source method - detective method — the two-source method
Work board (issues): [git.produktor.io/eSlider/2dph/issues](https://git.produktor.io/eSlider/2dph/issues).
PRs and CI: GitHub [`eSlider/2dph`](https://github.com/eSlider/2dph).
See [PLAN.md](PLAN.md) for decisions, execution status, and v2 open questions. See [PLAN.md](PLAN.md) for decisions, execution status, and v2 open questions.
-669
View File
@@ -1,669 +0,0 @@
package main
import (
"bufio"
"context"
"encoding/json"
"fmt"
"os"
"os/exec"
"path/filepath"
"regexp"
"strings"
"time"
)
type LinkedInMCPSource struct {
userDataDir string
limit int
}
type lnInboxItem struct {
ThreadID string `json:"thread_id"`
Participants string `json:"participants"`
LastMessage string `json:"last_message"`
LastMessageDate string `json:"last_message_date"`
Unread bool `json:"unread"`
}
// mcp-server-linkedin v4.22 returns get_inbox / get_conversation as
// {url, sections:{inbox|conversation: textblob}, references:{...}}.
// The conversation list lives in references (kind=conversation); messages live
// in the sections text blob, delimited by "<From> sent the following message
// at <time>" markers. See testdata/linkedin_*.json for the wire shape.
type lnEnvelope struct {
URL string `json:"url"`
Sections map[string]any `json:"sections"`
References map[string]any `json:"references"`
}
type lnReference struct {
Kind string `json:"kind"`
URL string `json:"url"`
Text string `json:"text"`
Context string `json:"context"`
}
type lnMessage struct {
From string `json:"from"`
Date string `json:"date"`
Text string `json:"text"`
}
var (
lnWeekdays = map[string]time.Weekday{
"SUNDAY": time.Sunday, "MONDAY": time.Monday, "TUESDAY": time.Tuesday,
"WEDNESDAY": time.Wednesday, "THURSDAY": time.Thursday,
"FRIDAY": time.Friday, "SATURDAY": time.Saturday,
}
lnMsgStartRe = regexp.MustCompile(`^(.+?) sent the following messages? at (.+)$`)
lnTimeRe = regexp.MustCompile(`\d{1,2}:\d{2}\s*[AP]M`)
)
func isWeekdayLine(s string) bool {
if _, ok := lnWeekdays[s]; ok {
return true
}
switch s {
case "TODAY", "YESTERDAY", "THIS WEEK", "LAST WEEK":
return true
}
return lnMonthDayRe.MatchString(s)
}
var lnMonthDayRe = regexp.MustCompile(`^[A-Z]{3}\s+\d{1,2}$`)
// parseLinkedInInbox extracts conversations from a get_inbox response.
func parseLinkedInInbox(text string) []lnInboxItem {
var env lnEnvelope
if err := json.Unmarshal([]byte(text), &env); err != nil {
return nil
}
refs, _ := env.References["inbox"].([]any)
var items []lnInboxItem
for _, r := range refs {
rr, ok := r.(map[string]any)
if !ok {
continue
}
if rr["kind"] != "conversation" {
continue
}
u, _ := rr["url"].(string)
tid := threadIDFromURL(u)
if !validThreadID(tid) {
continue
}
name, _ := rr["text"].(string)
items = append(items, lnInboxItem{
ThreadID: tid,
Participants: name,
})
}
return items
}
// parseLinkedInConversation parses the sections.conversation text blob into
// messages. Messages are delimited by "<From> sent the following message(s) at
// <time>" lines; each message body runs until the next marker. Day headers
// (all-caps weekdays) provide date context; times are mapped to the most
// recent matching weekday.
func parseLinkedInConversation(text string) []lnMessage {
var env lnEnvelope
if err := json.Unmarshal([]byte(text), &env); err != nil {
return nil
}
blob, _ := env.Sections["conversation"].(string)
if blob == "" {
return nil
}
var msgs []lnMessage
var cur *lnMessage
var body []string
day := ""
flush := func() {
if cur == nil {
return
}
cur.Text = strings.TrimSpace(strings.Join(body, "\n"))
if ts := linkedInTimestamp(day, cur.Date); ts != "" {
cur.Date = ts
}
if cur.Text != "" {
msgs = append(msgs, *cur)
}
cur = nil
body = nil
}
for _, raw := range strings.Split(blob, "\n") {
line := strings.TrimSpace(raw)
if line == "" {
continue
}
if isWeekdayLine(line) {
if line != day {
// A new day header terminates the previous message,
// which must keep the earlier date context.
flush()
}
day = line
continue
}
if m := lnMsgStartRe.FindStringSubmatch(line); m != nil {
flush()
cur = &lnMessage{From: strings.TrimSpace(m[1]), Date: strings.TrimSpace(m[2])}
continue
}
if cur == nil {
continue
}
// Skip "View X's profile" and the "<From> (pronouns) <time>" header.
if strings.HasPrefix(line, "View ") && strings.HasSuffix(line, "'s profile") {
continue
}
if strings.HasPrefix(line, cur.From) && lnTimeRe.MatchString(line) {
continue
}
body = append(body, line)
}
flush()
return msgs
}
// linkedInTimestamp maps a weekday, relative, or MON DD date header + clock
// string to a timestamp, or returns "" when the clock cannot be parsed.
func linkedInTimestamp(day, clock string) string {
t, err := time.Parse("3:04 PM", clock)
if err != nil {
return ""
}
now := time.Now()
var d time.Time
if wd, ok := lnWeekdays[day]; ok {
diff := (int(now.Weekday()) - int(wd) + 7) % 7
d = now.AddDate(0, 0, -diff)
} else {
switch day {
case "TODAY":
d = now
case "YESTERDAY":
d = now.AddDate(0, 0, -1)
case "THIS WEEK":
diff := int(now.Weekday())
d = now.AddDate(0, 0, -diff)
case "LAST WEEK":
diff := int(now.Weekday()) + 7
d = now.AddDate(0, 0, -diff)
default:
if m := lnMonthDayRe.FindStringSubmatch(day); m != nil {
// MON DD without a year: resolve to the most recent
// occurrence that is not in the future.
d = monthDayDate(day, now)
if d.IsZero() {
return t.Format("15:04")
}
} else {
// No date context; keep bare clock time.
return t.Format("15:04")
}
}
}
res := time.Date(d.Year(), d.Month(), d.Day(), t.Hour(), t.Minute(), 0, 0, time.UTC)
return res.UTC().Format(time.RFC3339)
}
var lnMonths = map[string]time.Month{
"JAN": time.January, "FEB": time.February, "MAR": time.March,
"APR": time.April, "MAY": time.May, "JUN": time.June,
"JUL": time.July, "AUG": time.August, "SEP": time.September,
"OCT": time.October, "NOV": time.November, "DEC": time.December,
}
// monthDayDate resolves "MON DD" to the most recent occurrence of that date,
// preferring the current year and falling back to the previous year when the
// date is in the future. Returns zero time when unresolvable.
func monthDayDate(day string, now time.Time) time.Time {
parts := strings.Fields(day)
if len(parts) != 2 {
return time.Time{}
}
mo, ok := lnMonths[parts[0]]
if !ok {
return time.Time{}
}
var dd int
if _, err := fmt.Sscanf(parts[1], "%d", &dd); err != nil {
return time.Time{}
}
if dd < 1 || dd > 31 {
return time.Time{}
}
d := time.Date(now.Year(), mo, dd, 0, 0, 0, 0, time.UTC)
if d.After(now) {
d = d.AddDate(-1, 0, 0)
}
if d.After(now) {
return time.Time{}
}
return d
}
func threadIDFromURL(u string) string {
u = strings.TrimSuffix(u, "/")
idx := strings.LastIndex(u, "/")
if idx < 0 {
return ""
}
return u[idx+1:]
}
// validThreadID rejects path segments that are not real thread ids (e.g. the
// literal "thread" or an empty trailing segment).
func validThreadID(id string) bool {
if id == "" || id == "thread" {
return false
}
return true
}
func NewLinkedInMCPSource(userDataDir string) *LinkedInMCPSource {
return &LinkedInMCPSource{userDataDir: userDataDir}
}
func (s *LinkedInMCPSource) Name() string { return "linkedin" }
func (s *LinkedInMCPSource) Sync(ctx context.Context, outDir string, limit int) error {
if limit > 0 {
s.limit = limit
}
// getConversation fetches one thread, recreating the MCP server when it
// wedges. A single 429 makes mcp-server-linkedin close its browser and
// refuse every later call ("still has a browser open"), so a broken server
// must be restarted rather than hammered.
getConversation := func(threadID string) ([]lnMessage, error) {
client, err := newLinkedInMCP(ctx, s.userDataDir)
if err != nil {
return nil, fmt.Errorf("linkedin mcp: %w", err)
}
defer client.Close()
msgs, err := client.GetConversation(ctx, "", threadID, msgLimitFor(s.limit))
if err != nil && wedged(err) {
fmt.Fprintf(os.Stderr, "chats: %s: server wedged, restarting broker\n", threadID)
time.Sleep(5 * time.Second)
client2, cerr := newLinkedInMCP(ctx, s.userDataDir)
if cerr == nil {
defer client2.Close()
msgs, err = client2.GetConversation(ctx, "", threadID, msgLimitFor(s.limit))
}
}
return msgs, err
}
client, err := newLinkedInMCP(ctx, s.userDataDir)
if err != nil {
return fmt.Errorf("linkedin mcp: %w", err)
}
inbox, err := client.GetInbox(ctx, 50)
if err != nil {
client.Close()
return fmt.Errorf("get_inbox: %w", err)
}
client.Close()
if len(inbox) == 0 {
fmt.Println("chats: no LinkedIn conversations found")
return nil
}
fmt.Printf("chats: found %d LinkedIn conversations\n", len(inbox))
for _, conv := range inbox {
convID := sanitizeDir(conv.ThreadID)
if convID == "" {
convID = fmt.Sprintf("conv_%d", time.Now().UnixNano())
}
parts := strings.SplitN(conv.Participants, ",", 2)
chatName := strings.TrimSpace(parts[0])
if chatName == "" {
chatName = convID
}
chatDir := filepath.Join(outDir, "linkedin", convID)
if err := os.MkdirAll(chatDir, 0755); err != nil {
fmt.Fprintf(os.Stderr, "chats: mkdir %s: %v\n", chatDir, err)
continue
}
msgs, err := getConversation(conv.ThreadID)
if err != nil {
fmt.Fprintf(os.Stderr, "chats: get_conversation %s: %v\n", convID, err)
continue
}
jsonlPath := filepath.Join(chatDir, "messages.jsonl")
// A rate-limited response can parse to zero messages. Never clobber
// previously synced data with an empty file.
if len(msgs) == 0 {
fmt.Fprintf(os.Stderr, "chats: %s (%s): 0 messages parsed, keeping existing file\n", chatName, convID)
continue
}
f, err := os.Create(jsonlPath)
if err != nil {
fmt.Fprintf(os.Stderr, "chats: create %s: %v\n", jsonlPath, err)
continue
}
enc := json.NewEncoder(f)
written := 0
for i, m := range msgs {
text := m.Text
if text == "" {
continue
}
ts := m.Date
if t, err := time.Parse("2006-01-02T15:04:05Z07:00", m.Date); err == nil {
ts = t.UTC().Format(time.RFC3339)
} else if t, err := time.Parse(time.RFC3339, m.Date); err == nil {
ts = t.UTC().Format(time.RFC3339)
}
chatMsg := Message{
ID: fmt.Sprintf("li_%s_%d", convID, i),
Timestamp: ts,
From: m.From,
Text: text,
Platform: "linkedin",
}
if err := enc.Encode(chatMsg); err != nil {
fmt.Fprintf(os.Stderr, "chats: encode: %v\n", err)
continue
}
written++
}
f.Close()
fmt.Printf("chats: synced %s (%s) — %d messages\n", chatName, convID, written)
// Pause between conversations to reduce LinkedIn rate limiting.
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(2 * time.Second):
}
}
return nil
}
type linkedInMCPClient struct {
cmd *exec.Cmd
stdin *bufio.Writer
stdout *bufio.Scanner
msgID int
}
func newLinkedInMCP(ctx context.Context, userDataDir string) (*linkedInMCPClient, error) {
args := []string{
"mcp-server-linkedin@latest",
"--user-data-dir", userDataDir,
"--no-auto-import",
"--no-daemon",
"--transport", "stdio",
"--login-timeout", "10",
"--browser-wait", "1",
"--browser-idle-timeout", "10",
"--log-level", "ERROR",
}
cmd := exec.CommandContext(ctx, "uvx", args...)
cmd.Env = os.Environ()
stdin, err := cmd.StdinPipe()
if err != nil {
return nil, fmt.Errorf("stdin pipe: %w", err)
}
stdout, err := cmd.StdoutPipe()
if err != nil {
return nil, fmt.Errorf("stdout pipe: %w", err)
}
cmd.Stderr = os.Stderr
if err := cmd.Start(); err != nil {
return nil, fmt.Errorf("start: %w", err)
}
c := &linkedInMCPClient{
cmd: cmd,
stdin: bufio.NewWriter(stdin),
stdout: bufio.NewScanner(stdout),
msgID: 0,
}
c.stdout.Buffer(make([]byte, 1<<20), 1<<20)
if err := c.initialize(ctx); err != nil {
c.Close()
return nil, fmt.Errorf("init: %w", err)
}
return c, nil
}
func (c *linkedInMCPClient) nextID() int {
c.msgID++
return c.msgID
}
func (c *linkedInMCPClient) initialize(ctx context.Context) error {
params := map[string]interface{}{
"protocolVersion": "2024-11-05",
"capabilities": map[string]interface{}{},
"clientInfo": map[string]string{
"name": "chats-sync",
"version": "0.1.0",
},
}
_, err := c.send(ctx, "initialize", params)
return err
}
func (c *linkedInMCPClient) send(ctx context.Context, method string, params interface{}) (json.RawMessage, error) {
id := c.nextID()
req := map[string]interface{}{
"jsonrpc": "2.0",
"id": id,
"method": method,
}
if params != nil {
req["params"] = params
}
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("marshal: %w", err)
}
if _, err := c.stdin.Write(body); err != nil {
return nil, fmt.Errorf("write: %w", err)
}
if err := c.stdin.WriteByte('\n'); err != nil {
return nil, fmt.Errorf("newline: %w", err)
}
if err := c.stdin.Flush(); err != nil {
return nil, fmt.Errorf("flush: %w", err)
}
for c.stdout.Scan() {
line := c.stdout.Text()
if line == "" {
continue
}
var resp struct {
JSONRPC string `json:"jsonrpc"`
ID int `json:"id"`
Result json.RawMessage `json:"result,omitempty"`
Error *struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error,omitempty"`
}
if err := json.Unmarshal([]byte(line), &resp); err != nil {
return nil, fmt.Errorf("unmarshal: %w\nline: %s", err, line[:min(len(line), 500)])
}
if resp.Error != nil {
return nil, fmt.Errorf("rpc error %d: %s", resp.Error.Code, resp.Error.Message)
}
return resp.Result, nil
}
return nil, fmt.Errorf("no response: %w", c.stdout.Err())
}
// msgLimitFor returns the per-conversation message cap for a sync.
func msgLimitFor(limit int) int {
if limit > 0 {
return limit
}
return 100
}
// wedged reports whether a conversation fetch failure means the MCP server
// closed its browser and will refuse every later call.
func wedged(err error) bool {
return strings.Contains(err.Error(), "still has a browser open")
}
// callTool invokes an MCP tool, retrying transient (rate-limit) failures.
func (c *linkedInMCPClient) callTool(ctx context.Context, name string, params map[string]interface{}) (json.RawMessage, error) {
var lastErr error
for attempt := 0; attempt < 3; attempt++ {
if attempt > 0 {
delay := time.Duration(1<<uint(attempt)) * 5 * time.Second
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-time.After(delay):
}
}
result, err := c.send(ctx, "tools/call", map[string]interface{}{
"name": name,
"arguments": params,
})
if err == nil {
// Tool-level errors surface as a successful RPC with an
// isError=true content entry.
if hint := toolErrorHint(result); hint != "" {
lastErr = fmt.Errorf("%s error: %s", name, hint)
if !isTransientLinkedInError(lastErr.Error()) {
return nil, lastErr
}
continue
}
return result, nil
}
lastErr = err
if !isTransientLinkedInError(err.Error()) {
return nil, err
}
}
return nil, fmt.Errorf("%s: %w", name, lastErr)
}
// toolErrorHint returns the tool's error text when the result has isError set.
func toolErrorHint(result json.RawMessage) string {
var toolRes struct {
Content []struct {
Type string `json:"type"`
Text string `json:"text"`
} `json:"content"`
IsError bool `json:"isError"`
}
if err := json.Unmarshal(result, &toolRes); err != nil || !toolRes.IsError {
return ""
}
if len(toolRes.Content) > 0 {
return toolRes.Content[0].Text
}
return "unknown tool error"
}
// isTransientLinkedInError reports whether a fetch failed due to rate limiting
// or a transient server error, which may succeed on retry.
func isTransientLinkedInError(msg string) bool {
return strings.Contains(msg, "503") || strings.Contains(msg, "429") ||
strings.Contains(msg, "ERR_HTTP_RESPONSE_CODE_FAILURE") ||
strings.Contains(msg, "ERR_ABORTED") ||
strings.Contains(msg, "Error calling tool") ||
strings.Contains(msg, "Unexpected error") ||
strings.Contains(msg, "still has a browser open")
}
func (c *linkedInMCPClient) GetInbox(ctx context.Context, limit int) ([]lnInboxItem, error) {
params := map[string]interface{}{
"limit": limit,
}
result, err := c.callTool(ctx, "get_inbox", params)
if err != nil {
return nil, err
}
var toolRes struct {
Content []struct {
Type string `json:"type"`
Text string `json:"text"`
} `json:"content"`
IsError bool `json:"isError"`
}
if err := json.Unmarshal(result, &toolRes); err != nil {
return nil, fmt.Errorf("unmarshal tool: %w", err)
}
if len(toolRes.Content) == 0 {
return nil, nil
}
text := toolRes.Content[0].Text
return parseLinkedInInbox(text), nil
}
func (c *linkedInMCPClient) GetConversation(ctx context.Context, username, threadID string, limit int) ([]lnMessage, error) {
params := map[string]interface{}{
"linkedin_username": username,
"thread_id": threadID,
"index": limit,
}
result, err := c.callTool(ctx, "get_conversation", params)
if err != nil {
return nil, err
}
var toolRes struct {
Content []struct {
Type string `json:"type"`
Text string `json:"text"`
} `json:"content"`
IsError bool `json:"isError"`
}
if err := json.Unmarshal(result, &toolRes); err != nil {
return nil, fmt.Errorf("unmarshal tool: %w", err)
}
if len(toolRes.Content) == 0 {
return nil, nil
}
text := toolRes.Content[0].Text
return parseLinkedInConversation(text), nil
}
func (c *linkedInMCPClient) Close() error {
if c.stdin != nil {
c.stdin.Flush()
}
if c.cmd != nil && c.cmd.Process != nil {
c.cmd.Process.Kill()
}
return nil
}
-215
View File
@@ -1,215 +0,0 @@
package main
import (
"errors"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
func readFixture(t *testing.T, name string) string {
t.Helper()
data, err := os.ReadFile(filepath.Join("testdata", name))
if err != nil {
t.Fatal(err)
}
return string(data)
}
// TestParseLinkedInInbox verifies get_inbox parsing against the v4.22 wire
// format (testdata/linkedin_inbox.json — synthetic Alice/Bob/Charlie).
func TestParseLinkedInInbox(t *testing.T) {
text := readFixture(t, "linkedin_inbox.json")
items := parseLinkedInInbox(text)
if len(items) != 2 {
t.Fatalf("expected 2 conversations (empty thread url skipped), got %d", len(items))
}
first := items[0]
if first.ThreadID == "" {
t.Error("expected thread id extracted from reference url")
}
if first.Participants != "Alice Example" {
t.Errorf("participants=%q, want Alice Example", first.Participants)
}
if !strings.HasPrefix(first.ThreadID, "2-") {
t.Errorf("unexpected thread id format %q", first.ThreadID)
}
if items[1].Participants != "Charlie Example" {
t.Errorf("second participant=%q, want Charlie Example", items[1].Participants)
}
}
// TestParseLinkedInInboxBadJSON verifies a non-JSON response yields no items
// rather than a panic or error.
func TestParseLinkedInInboxBadJSON(t *testing.T) {
if got := parseLinkedInInbox("Session expired"); len(got) != 0 {
t.Fatalf("expected no items for non-JSON, got %d", len(got))
}
}
// TestParseLinkedInConversation verifies message extraction from the sections
// blob (testdata/linkedin_conversation.json — synthetic Alice/Bob).
func TestParseLinkedInConversation(t *testing.T) {
text := readFixture(t, "linkedin_conversation.json")
msgs := parseLinkedInConversation(text)
if len(msgs) != 2 {
t.Fatalf("expected 2 messages, got %d", len(msgs))
}
if msgs[0].From != "Alice Example" {
t.Errorf("from=%q, want Alice Example", msgs[0].From)
}
if !strings.Contains(msgs[0].Text, "Senior Software Engineer") {
t.Errorf("alice text missing role, got %q", msgs[0].Text)
}
if msgs[0].Date == "" {
t.Error("expected message date")
}
if msgs[1].From != "Bob Example" {
t.Errorf("from=%q, want Bob Example", msgs[1].From)
}
}
// TestParseLinkedInConversationEmpty verifies empty/non-JSON blobs parse to
// zero messages.
func TestParseLinkedInConversationEmpty(t *testing.T) {
if got := parseLinkedInConversation("no data here"); len(got) != 0 {
t.Fatalf("expected 0 messages, got %d", len(got))
}
}
// TestLinkedInTimestamp verifies weekday+clock resolution to a recent UTC date.
func TestLinkedInTimestamp(t *testing.T) {
// The most recent Wednesday before/equal to "now".
ts := linkedInTimestamp("WEDNESDAY", "10:02 AM")
parsed, err := time.Parse(time.RFC3339, ts)
if err != nil {
t.Fatalf("unparseable timestamp %q: %v", ts, err)
}
if parsed.Weekday() != time.Wednesday {
t.Errorf("expected Wednesday, got %s", parsed.Weekday())
}
if parsed.Hour() != 10 || parsed.Minute() != 2 {
t.Errorf("expected 10:02, got %02d:%02d", parsed.Hour(), parsed.Minute())
}
now := time.Now()
diff := now.Sub(parsed)
if diff < 0 || diff > 7*24*time.Hour {
t.Errorf("timestamp %s is not within the last week of %s", parsed, now)
}
if got := linkedInTimestamp("MONDAY", "garbage"); got != "" {
t.Errorf("expected empty for bad clock, got %q", got)
}
if got := linkedInTimestamp("", "1:22 PM"); got != "13:22" {
t.Errorf("expected bare 13:22 for missing weekday, got %q", got)
}
// Relative day headers must resolve to full dates, not bare clocks.
today := linkedInTimestamp("TODAY", "9:42 AM")
yp, err := time.Parse(time.RFC3339, today)
if err != nil {
t.Fatalf("TODAY unparseable %q: %v", today, err)
}
if yp.Year() != now.Year() || yp.Month() != now.Month() || yp.Day() != now.Day() {
t.Errorf("TODAY expected %v, got %v", now, yp)
}
yest := linkedInTimestamp("YESTERDAY", "3:00 PM")
yp, err = time.Parse(time.RFC3339, yest)
if err != nil {
t.Fatalf("YESTERDAY unparseable %q: %v", yest, err)
}
if yp.Day() != now.AddDate(0, 0, -1).Day() {
t.Errorf("YESTERDAY expected day %d, got %d", now.AddDate(0, 0, -1).Day(), yp.Day())
}
// MON DD header (e.g. "JUN 25"): must resolve to a full date. The
// timestamp should fall within the current year (falling back to the
// prior year if the date would be in the future).
md := linkedInTimestamp("JUN 25", "10:48 AM")
mp, err := time.Parse(time.RFC3339, md)
if err != nil {
t.Fatalf("MON DD unparseable %q: %v", md, err)
}
if mp.Year() != now.Year() && mp.Year() != now.Year()-1 {
t.Errorf("JUN 25 expected year %d or %d, got %d", now.Year(), now.Year()-1, mp.Year())
}
if mp.Month() != time.June || mp.Day() != 25 {
t.Errorf("JUN 25 expected Jun 25, got %s %d", mp.Month(), mp.Day())
}
if mp.After(now) {
t.Errorf("JUN 25 resolved to the future: %s > %s", mp, now)
}
}
// TestTransientLinkedInError verifies rate-limit errors are retryable but
// genuine failures are not.
func TestTransientLinkedInError(t *testing.T) {
retryable := []string{
"get_conversation error: Error calling tool 'get_conversation'",
"get_conversation error: Unexpected error in get_conversation: net::ERR_HTTP_RESPONSE_CODE_FAILURE",
"rpc error 503: rate limited",
"rpc error 429: too many requests",
"get_conversation: get_conversation error: This server still has a browser open on the profile.",
}
for _, msg := range retryable {
if !isTransientLinkedInError(msg) {
t.Errorf("expected %q to be transient", msg)
}
}
permanent := []string{
"get_inbox error: bad credentials",
"rpc error -32602: Invalid request parameters",
"unmarshal: unexpected end of JSON input",
}
for _, msg := range permanent {
if isTransientLinkedInError(msg) {
t.Errorf("expected %q to be permanent", msg)
}
}
}
// TestMsgLimitFor verifies the per-conversation message cap resolution.
func TestMsgLimitFor(t *testing.T) {
if got := msgLimitFor(0); got != 100 {
t.Errorf("expected default 100, got %d", got)
}
if got := msgLimitFor(5); got != 5 {
t.Errorf("expected 5, got %d", got)
}
}
// TestWedged verifies the browser-open failure is recognized as a wedge.
func TestWedged(t *testing.T) {
if !wedged(errors.New("get_conversation error: This server still has a browser open on the profile")) {
t.Error("expected wedged error to be recognized")
}
if wedged(errors.New("get_conversation error: bad thing")) {
t.Error("unexpected wedge detection")
}
}
// TestThreadIDFromURL verifies thread id extraction.
func TestThreadIDFromURL(t *testing.T) {
cases := []struct {
url, want string
}{
{"/messaging/thread/2-abc123/", "2-abc123"},
{"/messaging/thread/2-abc123", "2-abc123"},
{"", ""},
{"/messaging/thread/", ""},
}
for _, c := range cases {
got := threadIDFromURL(c.url)
if c.want == "" && validThreadID(got) {
t.Errorf("threadIDFromURL(%q) = %q, want empty", c.url, got)
}
if c.want != "" && got != c.want {
t.Errorf("threadIDFromURL(%q) = %q, want %q", c.url, got, c.want)
}
}
}
+2 -1
View File
@@ -40,7 +40,8 @@ func main() {
fmt.Fprintf(os.Stderr, "chats: WhatsApp not implemented yet\n") fmt.Fprintf(os.Stderr, "chats: WhatsApp not implemented yet\n")
os.Exit(1) os.Exit(1)
case "linkedin": case "linkedin":
os.Exit(runSyncLinkedIn(platformArgs)) fmt.Fprintf(os.Stderr, "chats: LinkedIn not implemented yet\n")
os.Exit(1)
default: default:
fmt.Fprintf(os.Stderr, "chats: unknown platform %q\n", platform) fmt.Fprintf(os.Stderr, "chats: unknown platform %q\n", platform)
os.Exit(2) os.Exit(2)
-155
View File
@@ -1,155 +0,0 @@
#!/usr/bin/env python3
"""chats/refresh-linkedin-session - refresh LinkedIn MCP session from webtop CDP.
bin/chats/refresh-linkedin-session [--cdp URL] [--root DIR]
"""
Reads the current LinkedIn cookies out of the running Thorium browser in the
work-webtop container via CDP (Network.getAllCookies), copies the live browser
profile onto the source profile directory, and rewrites the portable
cookies.json + source-state.json that mcp-server-linkedin requires.
Usage:
refresh-linkedin-session [--cdp http://127.0.0.1:9222] [--root /var/tmp/liprofile]
[--container work-webtop] [--profile thorium-profile]
After the headless driver uses a copied profile, LinkedIn rotates the session
in that copy, so this must run before every sync.
"""
import asyncio
import json
import os
import shutil
import subprocess
import sys
import tempfile
import urllib.request
import websockets
def cdp_tab(ws_json):
for t in ws_json:
if t.get("webSocketDebuggerUrl"):
return t["webSocketDebuggerUrl"]
return None
async def get_cookies(ws_url):
async with websockets.connect(ws_url, max_size=50_000_000) as ws:
await ws.send(json.dumps({"id": 1, "method": "Network.getAllCookies", "params": {}}))
resp = await ws.recv()
return json.loads(resp).get("result", {}).get("cookies", [])
def write_source_state(root, profile_dir):
# Reuse the linkedin-mcp-server session_state module to write a valid
# source-state.json (same schema the daemon reads).
try:
from linkedin_mcp_server.session_state import canonical, write_source_state
write_source_state(canonical(__import__("pathlib").Path(profile_dir)))
return
except Exception:
pass
# Fallback: minimal schema-compatible state.
import uuid
state = {
"version": 1,
"source_runtime_id": "linux-amd64-host",
"login_generation": str(uuid.uuid4()),
"created_at": None,
"profile_path": profile_dir,
"cookies_path": os.path.join(root, "cookies.json"),
}
from datetime import datetime, timezone
state["created_at"] = datetime.now(timezone.utc).isoformat()
with open(os.path.join(root, "source-state.json"), "w") as f:
json.dump(state, f, indent=2)
def main():
args = sys.argv[1:]
cdp = "http://127.0.0.1:9222"
root = "/var/tmp/liprofile"
container = "work-webtop"
cprofile = "thorium-profile"
for i in range(0, len(args), 2):
k = args[i]
v = args[i + 1] if i + 1 < len(args) else ""
if k == "--cdp":
cdp = v
elif k == "--root":
root = v
elif k == "--container":
container = v
elif k == "--profile":
cprofile = v
profile_dir = os.path.join(root, "profile")
os.makedirs(profile_dir, exist_ok=True)
# 1. Clear stale daemon/browser locks so the server can claim the profile.
for lock in ("profile-claim.lock", "profile.lock", "daemon.lock", "lease.lock"):
p = os.path.join(root, lock)
if os.path.exists(p):
os.remove(p)
for name in os.listdir(profile_dir):
if name.startswith("Singleton"):
os.remove(os.path.join(profile_dir, name))
for name in os.listdir(root):
if name.startswith("invalid-state-"):
shutil.rmtree(os.path.join(root, name), ignore_errors=True)
# 1. Copy the live browser profile (cookies DB + Local State) so the
# session the driver launches carries the current login.
subprocess.run(
["docker", "cp", f"{container}:/config/{cprofile}/Default", os.path.join(profile_dir, "Default")],
check=True, capture_output=True,
)
subprocess.run(
["docker", "cp", f"{container}:/config/{cprofile}/Local State", os.path.join(profile_dir, "Local State")],
check=True, capture_output=True,
)
for lock in ("SingletonLock", "SingletonCookie", "SingletonSocket"):
p = os.path.join(profile_dir, lock)
if os.path.exists(p):
os.remove(p)
# 2. Pull the live cookies out of the running browser.
with urllib.request.urlopen(f"{cdp}/json", timeout=5) as r:
tabs = json.loads(r.read())
ws_url = cdp_tab(tabs)
if not ws_url:
sys.stderr.write("refresh-linkedin-session: no CDP tab\n")
sys.exit(1)
cookies = asyncio.run(get_cookies(ws_url))
li = [c for c in cookies if "linkedin" in c.get("domain", "")]
out = []
for c in li:
domain = c.get("domain", "")
if domain in (".www.linkedin.com", "www.linkedin.com"):
domain = ".linkedin.com"
out.append({
"name": c["name"],
"value": c["value"].strip('"'),
"domain": domain,
"path": c.get("path", "/"),
"expires": c.get("expires", -1),
"httpOnly": c.get("httpOnly", False),
"secure": c.get("secure", False),
"sameSite": c.get("sameSite", "None"),
})
with open(os.path.join(root, "cookies.json"), "w") as f:
json.dump(out, f, indent=2)
write_source_state(root, profile_dir)
sys.stderr.write(f"refresh-linkedin-session: {len(out)} cookies, profile refreshed\n")
if __name__ == "__main__":
main()
-105
View File
@@ -1,105 +0,0 @@
package main
import (
"context"
"flag"
"fmt"
"os"
"os/exec"
"path/filepath"
"time"
)
func checkLinkedInSession(userDataDir string) (bool, error) {
// Validate the source-session files without launching a browser. A full
// `--status` run spawns Chromium and loads /feed/, doubling the automation
// exposed to LinkedIn (429 rate limits) before the sync even starts.
root := filepath.Dir(userDataDir)
sessionFiles := []string{
filepath.Join(root, "source-state.json"),
filepath.Join(root, "cookies.json"),
filepath.Join(userDataDir, "Default", "Cookies"),
}
for _, f := range sessionFiles {
if _, err := os.Stat(f); err != nil {
return true, fmt.Errorf("missing session file %s", f)
}
}
return false, nil
}
func runSyncLinkedIn(args []string) int {
fs := flag.NewFlagSet("chats sync linkedin", flag.ContinueOnError)
limit := fs.Int("limit", 0, "max messages per conversation (0 = all)")
refresh := fs.Bool("refresh", false, "refresh session from live webtop browser before sync")
help := fs.Bool("help", false, "")
fs.SetOutput(os.Stderr)
if err := fs.Parse(args); err != nil {
return 2
}
if *help {
fmt.Fprintln(os.Stderr, "usage: chats sync linkedin [--limit N] [--refresh]")
return 0
}
userDataDir := envVar("LINKEDIN_USER_DATA_DIR", "")
if userDataDir == "" {
home, _ := os.UserHomeDir()
userDataDir = home + "/.linkedin-mcp/profile"
}
if *refresh {
if code := refreshLinkedInSession(userDataDir); code != 0 {
return code
}
}
// Check session files first (no browser launch).
loginNeeded, err := checkLinkedInSession(userDataDir)
if err != nil {
fmt.Fprintf(os.Stderr, "chats: linkedin status check: %v\n", err)
}
if loginNeeded {
fmt.Fprintf(os.Stderr, "chats: LinkedIn session missing. Run:\n")
fmt.Fprintf(os.Stderr, " chats sync linkedin --refresh\n")
fmt.Fprintf(os.Stderr, "or point LINKEDIN_USER_DATA_DIR at a valid session\n")
return 1
}
src := NewLinkedInMCPSource(userDataDir)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel()
start := time.Now()
if err := src.Sync(ctx, chatsDir(), *limit); err != nil {
fmt.Fprintf(os.Stderr, "chats sync linkedin: %v\n", err)
return 1
}
fmt.Printf("chats sync linkedin: completed in %s\n", time.Since(start).Round(time.Millisecond))
return 0
}
// refreshLinkedInSession re-syncs the LinkedIn source session from the live
// webtop browser via the vendored refresh-linkedin-session helper.
func refreshLinkedInSession(userDataDir string) int {
exe, err := os.Executable()
if err != nil {
fmt.Fprintf(os.Stderr, "chats: resolve executable: %v\n", err)
return 1
}
helper := filepath.Join(filepath.Dir(exe), "refresh-linkedin-session")
if _, err := os.Stat(helper); err != nil {
// Fall back to the source tree helper next to this command file.
helper = "bin/chats/refresh-linkedin-session"
}
root := filepath.Dir(userDataDir)
cmd := exec.Command(helper, "--root", root)
cmd.Stdout = os.Stderr
cmd.Stderr = os.Stderr
if err := cmd.Run(); err != nil {
fmt.Fprintf(os.Stderr, "chats: linkedin session refresh: %v\n", err)
return 1
}
return 0
}
-12
View File
@@ -1,12 +0,0 @@
{
"url": "https://www.linkedin.com/messaging/thread/2-YWxpY2UtYm9iLXRocmVhZC0xMjM=/",
"sections": {
"conversation": "WEDNESDAY\nAlice Example sent the following message at 10:02 AM\nView Alice Example's profile\nAlice Example (She/Her) 10:02 AM\nHi Bob, we have a Senior Software Engineer role that matches your Go and Python background. Happy to share more if you are open to a chat.\n\nBob Example sent the following messages at 1:22 PM\nView Bob Example's profile\nBob Example 1:22 PM\nHi Alice, thanks for reaching out — yes, I am open to exploring a Senior Software Engineer role. Happy to do a short video call.\n"
},
"references": {
"conversation": [
{"kind": "person", "url": "/in/alice-example/", "text": "Alice Example"},
{"kind": "person", "url": "/in/bob-example/", "text": "Bob Example"}
]
}
}
-34
View File
@@ -1,34 +0,0 @@
{
"url": "https://www.linkedin.com/messaging/",
"sections": {
"inbox": "Messaging\nInbox\nConversation List\nAlice Example\nExciting opportunity for a senior software engineer\n"
},
"references": {
"inbox": [
{
"kind": "conversation",
"url": "/messaging/thread/2-YWxpY2UtYm9iLXRocmVhZC0xMjM=/",
"context": "inbox",
"text": "Alice Example"
},
{
"kind": "person",
"url": "/in/alice-example/",
"text": "Alice Example",
"context": "inbox"
},
{
"kind": "conversation",
"url": "/messaging/thread/2-Y2hhcmxpZS1ib2ItdGhyZWFkLTQ1Ng==/",
"context": "inbox",
"text": "Charlie Example"
},
{
"kind": "conversation",
"url": "/messaging/thread/",
"context": "inbox",
"text": "should-be-skipped"
}
]
}
}
+7 -18
View File
@@ -5,7 +5,6 @@ import (
"fmt" "fmt"
"os" "os"
"path/filepath" "path/filepath"
"strings"
lbug "github.com/LadybugDB/go-ladybug" lbug "github.com/LadybugDB/go-ladybug"
) )
@@ -45,10 +44,10 @@ func dbPath() string {
} }
func openBrain() error { func openBrain() error {
return openWithSandbox(eps()) return openWithOpts(2, eps())
} }
func openWithSandbox(epsv string) error { func openWithOpts(allow int, epsv string) error {
cfg := lbug.DefaultSystemConfig() cfg := lbug.DefaultSystemConfig()
cfg.MaxNumThreads = 8 cfg.MaxNumThreads = 8
cfg.BufferPoolSize = 1 << 30 // 1GB cfg.BufferPoolSize = 1 << 30 // 1GB
@@ -58,30 +57,20 @@ func openWithSandbox(epsv string) error {
if err != nil { if err != nil {
return fmt.Errorf("OpenDatabase: %w", err) return fmt.Errorf("OpenDatabase: %w", err)
} }
if epsv != "" {
if _, err := conn.Query("SET STREAM_SANDBOX = '" + epsv + "'"); err != nil {
return err
}
}
conn, err = lbug.OpenConnection(db) conn, err = lbug.OpenConnection(db)
if err != nil { if err != nil {
closeBrain()
return fmt.Errorf("OpenConnection: %w", err) return fmt.Errorf("OpenConnection: %w", err)
} }
// Session settings need a live connection; running this before
// OpenConnection dereferenced a nil *Connection.
if epsv != "" {
if strings.ContainsAny(epsv, "'\\") {
closeBrain()
return fmt.Errorf("SET STREAM_SANDBOX: invalid value")
}
if _, err := conn.Query("SET STREAM_SANDBOX = '" + epsv + "'"); err != nil {
closeBrain()
return fmt.Errorf("SET STREAM_SANDBOX: %w", err)
}
}
if _, err := conn.Query("LOAD EXTENSION FTS"); err != nil { if _, err := conn.Query("LOAD EXTENSION FTS"); err != nil {
closeBrain()
return fmt.Errorf("LOAD EXTENSION FTS: %w", err) return fmt.Errorf("LOAD EXTENSION FTS: %w", err)
} }
if _, err := conn.Query("LOAD EXTENSION VECTOR"); err != nil { if _, err := conn.Query("LOAD EXTENSION VECTOR"); err != nil {
closeBrain()
return fmt.Errorf("LOAD EXTENSION VECTOR: %w", err) return fmt.Errorf("LOAD EXTENSION VECTOR: %w", err)
} }
return nil return nil
+1 -3
View File
@@ -8,9 +8,7 @@
// kbsearch --list-model print the resolved model dir // kbsearch --list-model print the resolved model dir
// //
// The potion-multilingual model is loaded only in `serve`; a CLI reuses the // The potion-multilingual model is loaded only in `serve`; a CLI reuses the
// daemon over localhost HTTP (KBSEARCH_PORT, default 17830) and starts one in // daemon over localhost HTTP (falling back to in-process embedding).
// the background when none answers. KBSEARCH_NO_DAEMON=1 skips that and embeds
// in-process instead.
package main package main
import ( import (
-72
View File
@@ -1,72 +0,0 @@
package rank
import (
"fmt"
"strconv"
"strings"
)
const Usage = `usage: kbsearch "query" [--root facts|info] [--repo REPO] [-n N] [--json]
kbsearch serve [port]
kbsearch --list-model`
type Options struct {
Query string
Root string
Repo string
Limit int
JSONOut bool
ListModel bool
}
// ParseArgs reads flags. Unknown flags are an error: silently dropping them
// meant `--hop 1` vanished and its argument `1` was appended to the query.
// --hop is recognised so it cannot be swallowed; it is not implemented until
// File/FROM_FILE edges exist.
func ParseArgs(args []string) (Options, error) {
opt := Options{Limit: 20}
var queryArgs []string
for i := 0; i < len(args); i++ {
arg := args[i]
wantsValue := arg == "--root" || arg == "--repo" || arg == "-n" || arg == "--hop"
if wantsValue && i+1 >= len(args) {
return opt, fmt.Errorf("%s needs a value", arg)
}
switch arg {
case "--root":
i++
opt.Root = args[i]
if opt.Root != "facts" && opt.Root != "info" {
return opt, fmt.Errorf("--root must be facts or info, got %q", opt.Root)
}
case "--repo":
i++
opt.Repo = args[i]
case "-n":
i++
n, err := strconv.Atoi(args[i])
if err != nil || n < 1 {
return opt, fmt.Errorf("-n must be a positive integer, got %q", args[i])
}
opt.Limit = n
case "--hop":
return opt, fmt.Errorf("--hop is not implemented yet (needs File/FROM_FILE edges)")
case "--json":
opt.JSONOut = true
case "--list-model":
opt.ListModel = true
default:
if strings.HasPrefix(arg, "-") {
return opt, fmt.Errorf("unknown flag %q", arg)
}
queryArgs = append(queryArgs, arg)
}
}
opt.Query = strings.TrimSpace(strings.Join(queryArgs, " "))
if opt.Query == "" && !opt.ListModel {
return opt, fmt.Errorf("no query given")
}
return opt, nil
}
-9
View File
@@ -1,9 +0,0 @@
package rank
// BM25 ranks best-first, so the top hits are the *highest* scores; cosine
// distance ranks best-first ascending. Both mirror kblib.py.
const FTSStmt = "CALL QUERY_FTS_INDEX('Leaf', 'id', $q) " +
"RETURN node.id, node.text, node.root, node.source, score ORDER BY score DESC LIMIT $n"
const VecStmt = "CALL QUERY_VECTOR_INDEX('Leaf', 'Leaf_vec', $q, $n) " +
"RETURN node.id, node.text, node.root, node.source, distance ORDER BY distance LIMIT $n"
-100
View File
@@ -1,100 +0,0 @@
// Package rank is the cgo-free ranking and CLI parsing for kbsearch.
// CI can `go test ./rank` without the native ladybug library.
package rank
import (
"sort"
"strings"
)
// Hit is one search result, mirroring the python script's dict shape.
type Hit struct {
ID string `json:"id"`
Text string `json:"text"`
Root string `json:"root"`
Source string `json:"-"`
Score float64 `json:"score"`
Snippet string `json:"snippet,omitempty"`
}
// rrfK dampens the contribution of low ranks; same constant as kblib.py.
const rrfK = 60
// RankAndFilter fuses the two hit lists, applies --root/--repo, then cuts to
// limit. Cutting first dropped every matching leaf ranked below the cut, so
// `--root facts` came back empty whenever info leafs filled the top N.
// limit <= 0 keeps everything.
func RankAndFilter(fts, vec []Hit, root, repo string, limit int) []Hit {
out := Hybrid(fts, vec, 0)
if root != "" {
out = FilterRoot(out, root)
}
if repo != "" {
out = FilterRepo(out, repo)
}
if limit > 0 && len(out) > limit {
out = out[:limit]
}
return out
}
// Hybrid merges FTS and vector hits by reciprocal rank fusion.
// limit <= 0 returns the full fused list.
func Hybrid(fts, vec []Hit, limit int) []Hit {
byID := make(map[string]Hit, len(fts)+len(vec))
rrf := make(map[string]float64, len(fts)+len(vec))
for i, h := range fts {
byID[h.ID] = h
rrf[h.ID] += 1.0 / (rrfK + float64(i+1))
}
for i, h := range vec {
if existing, ok := byID[h.ID]; !ok {
byID[h.ID] = h
} else if existing.Score == 0 {
existing.Score = h.Score
byID[h.ID] = existing
}
rrf[h.ID] += 1.0 / (rrfK + float64(i+1))
}
ids := make([]string, 0, len(rrf))
for id := range rrf {
ids = append(ids, id)
}
sort.Slice(ids, func(i, j int) bool {
if rrf[ids[i]] != rrf[ids[j]] {
return rrf[ids[i]] > rrf[ids[j]]
}
return ids[i] < ids[j]
})
if limit > 0 && len(ids) > limit {
ids = ids[:limit]
}
out := make([]Hit, 0, len(ids))
for _, id := range ids {
out = append(out, byID[id])
}
return out
}
func FilterRoot(hits []Hit, root string) []Hit {
var out []Hit
for _, h := range hits {
if h.Root == root {
out = append(out, h)
}
}
return out
}
func FilterRepo(hits []Hit, repo string) []Hit {
var out []Hit
for _, h := range hits {
if strings.Contains(h.Source, repo) {
out = append(out, h)
}
}
return out
}
-143
View File
@@ -1,143 +0,0 @@
// Unit tests for ranking/filtering and CLI parsing (no db, no model, offline).
package rank
import (
"strings"
"testing"
)
func h(id, root, source string) Hit {
return Hit{ID: id, Text: id, Root: root, Source: source}
}
func ids(hits []Hit) []string {
out := make([]string, len(hits))
for i, hit := range hits {
out[i] = hit.ID
}
return out
}
func eq(t *testing.T, got []Hit, want ...string) {
t.Helper()
g := ids(got)
if len(g) != len(want) {
t.Fatalf("got %v, want %v", g, want)
}
for i := range want {
if g[i] != want[i] {
t.Fatalf("got %v, want %v", g, want)
}
}
}
// A facts leaf that ranks below the limit in the unfiltered list must still
// be returned for --root facts. Filtering after truncation loses it.
func TestRankAndFilterFiltersBeforeLimit(t *testing.T) {
fts := []Hit{
h("i1", "info", "docs/a.md"),
h("i2", "info", "docs/b.md"),
h("i3", "info", "docs/c.md"),
h("f1", "facts", "docker ps x compose"),
}
eq(t, RankAndFilter(fts, nil, "facts", "", 2), "f1")
}
func TestRankAndFilterRepoFiltersBeforeLimit(t *testing.T) {
fts := []Hit{
h("a", "info", "eSlider/2dph:README.md"),
h("b", "info", "eSlider/2dph:PLAN.md"),
h("c", "info", "eSlider/ops:compose.yaml"),
}
eq(t, RankAndFilter(fts, nil, "", "ops", 2), "c")
}
func TestRankAndFilterTruncatesToLimit(t *testing.T) {
fts := []Hit{h("a", "info", "x"), h("b", "info", "x"), h("c", "info", "x")}
eq(t, RankAndFilter(fts, nil, "", "", 2), "a", "b")
}
func TestRankAndFilterLimitZeroKeepsAll(t *testing.T) {
fts := []Hit{h("a", "info", "x"), h("b", "info", "x")}
eq(t, RankAndFilter(fts, nil, "", "", 0), "a", "b")
}
func TestHybridFusesBothRetrievers(t *testing.T) {
fts := []Hit{h("only-fts", "info", "x"), h("both", "info", "x")}
vec := []Hit{h("only-vec", "info", "x"), h("both", "info", "x")}
eq(t, Hybrid(fts, vec, 0), "both", "only-fts", "only-vec")
}
func TestHybridTiesAreDeterministic(t *testing.T) {
fts := []Hit{h("b", "info", "x"), h("a", "info", "x")}
first := ids(Hybrid(fts, nil, 0))
for i := 0; i < 50; i++ {
got := ids(Hybrid(fts, nil, 0))
for j := range first {
if got[j] != first[j] {
t.Fatalf("unstable order: %v then %v", first, got)
}
}
}
}
func TestHybridKeepsVectorScoreForSharedHit(t *testing.T) {
fts := []Hit{{ID: "x", Root: "info", Score: 0}}
vec := []Hit{{ID: "x", Root: "info", Score: 0.87}}
got := Hybrid(fts, vec, 0)
if len(got) != 1 || got[0].Score != 0.87 {
t.Fatalf("got %+v, want score 0.87", got)
}
}
// The old parser dropped unknown flags and appended their arguments to the
// query, so `search "q" --hop 1` searched for "q 1". --hop is not implemented
// here (needs File edges); it must still fail closed instead of changing q.
func TestParseHopIsNotSwallowedIntoTheQuery(t *testing.T) {
_, err := ParseArgs([]string{"what runs on arc-2", "--hop", "1"})
if err == nil {
t.Fatal("expected --hop to error (not implemented), not be swallowed")
}
if !strings.Contains(err.Error(), "--hop") {
t.Fatalf("error should name --hop, got %v", err)
}
}
func TestParseRejectsUnknownFlags(t *testing.T) {
if _, err := ParseArgs([]string{"query", "--nope"}); err == nil {
t.Fatal("unknown flag accepted")
}
}
func TestParseRejectsBadValues(t *testing.T) {
for _, args := range [][]string{
{"q", "-n", "zero"},
{"q", "-n", "0"},
{"q", "--root", "nonsense"},
{"q", "--hop"},
{"--json"},
} {
if _, err := ParseArgs(args); err == nil {
t.Errorf("accepted %v", args)
}
}
}
func TestParseDefaults(t *testing.T) {
opt, err := ParseArgs([]string{"two", "words", "--json"})
if err != nil || opt.Query != "two words" || opt.Limit != 20 || !opt.JSONOut {
t.Fatalf("got %+v err=%v", opt, err)
}
}
func TestListModelNeedsNoQuery(t *testing.T) {
if _, err := ParseArgs([]string{"--list-model"}); err != nil {
t.Fatalf("unexpected error: %v", err)
}
}
func TestFTSQueryOrdersByScoreDescending(t *testing.T) {
if !strings.Contains(FTSStmt, "ORDER BY score DESC") {
t.Fatalf("FTS query must order by score DESC, got:\n%s", FTSStmt)
}
}
+131 -25
View File
@@ -13,12 +13,12 @@ import (
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
"sort"
"strconv" "strconv"
"syscall" "strings"
"time" "time"
lbug "github.com/LadybugDB/go-ladybug" lbug "github.com/LadybugDB/go-ladybug"
"github.com/eSlider/2dph/bin/kbsearch/rank"
) )
const defaultPort = 17830 const defaultPort = 17830
@@ -26,15 +26,45 @@ const daemonPath = "/embed"
const healthPath = "/health" const healthPath = "/health"
func runSearch(args []string) int { func runSearch(args []string) int {
opt, err := rank.ParseArgs(args) // Manual flag parsing to allow flags after query (like Python argparse)
if err != nil { root := ""
fmt.Fprintf(os.Stderr, "kbsearch: %v\n%s\n", err, rank.Usage) repo := ""
return 2 limit := 20
} jsonOut := false
root, repo, limit, query := opt.Root, opt.Repo, opt.Limit, opt.Query listModel := false
jsonOut := opt.JSONOut
if opt.ListModel { var queryArgs []string
for i := 0; i < len(args); i++ {
switch args[i] {
case "--root":
if i+1 < len(args) {
root = args[i+1]
i++
}
case "--repo":
if i+1 < len(args) {
repo = args[i+1]
i++
}
case "-n":
if i+1 < len(args) {
if n, err := strconv.Atoi(args[i+1]); err == nil {
limit = n
}
i++
}
case "--json":
jsonOut = true
case "--list-model":
listModel = true
default:
if !strings.HasPrefix(args[i], "-") {
queryArgs = append(queryArgs, args[i])
}
}
}
if listModel {
dir, err := modelDir() dir, err := modelDir()
if err != nil { if err != nil {
fmt.Fprintln(os.Stderr, err) fmt.Fprintln(os.Stderr, err)
@@ -44,6 +74,12 @@ func runSearch(args []string) int {
return 0 return 0
} }
query := strings.TrimSpace(strings.Join(queryArgs, " "))
if query == "" {
fmt.Fprintln(os.Stderr, "usage: kbsearch \"query\" [--root facts|info] [--repo REPO] [-n N] [--json]")
return 1
}
if err := openBrain(); err != nil { if err := openBrain(); err != nil {
fmt.Fprintf(os.Stderr, "open brain: %v\n", err) fmt.Fprintf(os.Stderr, "open brain: %v\n", err)
return 1 return 1
@@ -67,7 +103,17 @@ func runSearch(args []string) int {
fmt.Fprintf(os.Stderr, "vec: %v\n", err) fmt.Fprintf(os.Stderr, "vec: %v\n", err)
} }
results := rank.RankAndFilter(fts, vec, root, repo, limit) results := hybrid(fts, vec, limit)
if root != "" {
results = filterRoot(results, root)
}
if repo != "" {
results = filterRepo(results, repo)
}
if len(results) > limit {
results = results[:limit]
}
for i := range results { for i := range results {
if results[i].Text != "" { if results[i].Text != "" {
@@ -104,7 +150,10 @@ func b2i(err error) int {
} }
func queryFTS(text string, limit int) ([]Hit, error) { func queryFTS(text string, limit int) ([]Hit, error) {
stmt, err := conn.Prepare(rank.FTSStmt) stmt, err := conn.Prepare(
"CALL QUERY_FTS_INDEX('Leaf', 'id', $q) " +
"RETURN node.id, node.text, node.root, node.source, score ORDER BY score LIMIT $n",
)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -121,7 +170,10 @@ func queryVector(emb []float64, limit int) ([]Hit, error) {
for i, v := range emb { for i, v := range emb {
embList[i] = v embList[i] = v
} }
stmt, err := conn.Prepare(rank.VecStmt) stmt, err := conn.Prepare(
"CALL QUERY_VECTOR_INDEX('Leaf', 'Leaf_vec', $q, $n) " +
"RETURN node.id, node.text, node.root, node.source, distance ORDER BY distance LIMIT $n",
)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -196,6 +248,70 @@ func toJSONOut(hits []Hit, query, rootFilter string) *jsonOut {
} }
} }
func hybrid(fts, vec []Hit, limit int) []Hit {
byID := make(map[string]Hit)
rrf := make(map[string]float64)
for rank, h := range fts {
byID[h.ID] = h
rrf[h.ID] += 1.0 / (60 + float64(rank+1))
}
for rank, h := range vec {
if _, ok := byID[h.ID]; !ok {
byID[h.ID] = h
} else {
existing := byID[h.ID]
if existing.Score == 0 {
existing.Score = h.Score
byID[h.ID] = existing
}
}
rrf[h.ID] += 1.0 / (60 + float64(rank+1))
}
type scored struct {
id string
rrf float64
}
var scoredList []scored
for id, v := range rrf {
scoredList = append(scoredList, scored{id, v})
}
sort.Slice(scoredList, func(i, j int) bool {
return scoredList[i].rrf > scoredList[j].rrf
})
var out []Hit
for i, s := range scoredList {
if i >= limit {
break
}
h := byID[s.id]
out = append(out, h)
}
return out
}
func filterRoot(hits []Hit, root string) []Hit {
var out []Hit
for _, h := range hits {
if h.Root == root {
out = append(out, h)
}
}
return out
}
func filterRepo(hits []Hit, repo string) []Hit {
var out []Hit
for _, h := range hits {
if strings.Contains(h.Source, repo) {
out = append(out, h)
}
}
return out
}
func resultsToDicts(hits []Hit) []any { func resultsToDicts(hits []Hit) []any {
out := make([]any, len(hits)) out := make([]any, len(hits))
for i, h := range hits { for i, h := range hits {
@@ -266,16 +382,10 @@ func embedQuery(text string) ([]float64, error) {
port = p port = p
} }
} }
if emb, err := tryDaemon(text, port); err == nil { emb, err := tryDaemon(text, port)
if err == nil {
return emb, nil return emb, nil
} }
if os.Getenv("KBSEARCH_NO_DAEMON") == "" {
if err := ensureDaemon(port); err == nil {
if emb, err := tryDaemon(text, port); err == nil {
return emb, nil
}
}
}
model, err := loadModel() model, err := loadModel()
if err != nil { if err != nil {
@@ -337,13 +447,9 @@ func ensureDaemon(port int) error {
cmd.Dir, _ = filepath.Split(self) cmd.Dir, _ = filepath.Split(self)
cmd.Stdout = nil cmd.Stdout = nil
cmd.Stderr = nil cmd.Stderr = nil
cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
if err := cmd.Start(); err != nil { if err := cmd.Start(); err != nil {
return err return err
} }
if cmd.Process != nil {
_ = cmd.Process.Release()
}
for i := 0; i < 40; i++ { for i := 0; i < 40; i++ {
time.Sleep(250 * time.Millisecond) time.Sleep(250 * time.Millisecond)
+10 -8
View File
@@ -1,14 +1,16 @@
// Common types and helpers for kbsearch. // Common types and helpers for kbsearch.
package main package main
import ( import "os"
"os"
"github.com/eSlider/2dph/bin/kbsearch/rank"
)
func eps() string { return os.Getenv("KBTEST_EPS") } func eps() string { return os.Getenv("KBTEST_EPS") }
// Hit is the search hit type; ranking lives in package rank so CI can test // Hit is one search result, mirroring the python script's dict shape.
// it without the native ladybug library. type Hit struct {
type Hit = rank.Hit ID string `json:"id"`
Text string `json:"text"`
Root string `json:"root"`
Source string `json:"-"` // for repo filtering, not in output
Score float64 `json:"score"`
Snippet string `json:"snippet,omitempty"`
}
+1 -3
View File
@@ -32,7 +32,6 @@ func ParseCLI(args []string) (CLIConfig, int, error) {
offset = fs.Int("offset", 0, "skip first N messages per source") offset = fs.Int("offset", 0, "skip first N messages per source")
force = fs.Bool("force", false, "overwrite existing message.json + attachments") force = fs.Bool("force", false, "overwrite existing message.json + attachments")
dryRun = fs.Bool("dry-run", false, "list message counts without writing") dryRun = fs.Bool("dry-run", false, "list message counts without writing")
query = fs.String("query", "in:inbox", "Gmail search query (gmail source only)")
srcs = fs.String("source", "onlyoffice", "comma list: onlyoffice,gmail (default onlyoffice)") srcs = fs.String("source", "onlyoffice", "comma list: onlyoffice,gmail (default onlyoffice)")
help = fs.Bool("help", false, "usage") help = fs.Bool("help", false, "usage")
) )
@@ -61,7 +60,6 @@ func ParseCLI(args []string) (CLIConfig, int, error) {
Offset: *offset, Offset: *offset,
Force: *force, Force: *force,
DryRun: *dryRun, DryRun: *dryRun,
Query: *query,
Policy: RetryPolicy{}, Policy: RetryPolicy{},
} }
cli := CLIConfig{Sync: cfg, Env: *env, Sources: *srcs} cli := CLIConfig{Sync: cfg, Env: *env, Sources: *srcs}
@@ -97,7 +95,7 @@ func Main(args []string) int {
return code return code
} }
if cli.Help { if cli.Help {
fmt.Fprintln(os.Stderr, "usage: bin/mail/sync.go [--source onlyoffice,gmail] [--query GMAIL_Q] [--limit N] [--offset N] [--workers N] [--force] [--dry-run]") fmt.Fprintln(os.Stderr, "usage: bin/mail/sync.go [--source onlyoffice,gmail] [--limit N] [--offset N] [--workers N] [--force] [--dry-run]")
return 0 return 0
} }
ctx, cancel := context.WithTimeout(context.Background(), 6*time.Hour) ctx, cancel := context.WithTimeout(context.Background(), 6*time.Hour)
+3 -16
View File
@@ -112,7 +112,6 @@ type SyncConfig struct {
Offset int // skip first N messages per source Offset int // skip first N messages per source
Force bool // overwrite existing message.json + attachments Force bool // overwrite existing message.json + attachments
DryRun bool // list without writing DryRun bool // list without writing
Query string // Gmail search query; default in:inbox
Policy RetryPolicy Policy RetryPolicy
} }
@@ -137,17 +136,9 @@ type ooSource struct {
c *OOClient c *OOClient
page int page int
} }
// gmailAPI is the Gmail client surface gmailSource needs. *GmailClient implements it.
type gmailAPI interface {
ListIDs(ctx context.Context, q string, maxIDs int, pageToken string) ([]string, string, error)
GetMessage(ctx context.Context, id string) (*Message, error)
DownloadAttachment(ctx context.Context, msgID, attID string) ([]byte, error)
}
type gmailSource struct { type gmailSource struct {
c gmailAPI c *GmailClient
cur string cur string
query string
} }
func (s *ooSource) Folder() string { return "inbox" } func (s *ooSource) Folder() string { return "inbox" }
@@ -180,11 +171,7 @@ func (s *ooSource) DownloadAttachment(ctx context.Context, msg *Message, att Att
} }
func (s *gmailSource) ListIDs(ctx context.Context, limit int, cursor string) ([]string, string, error) { func (s *gmailSource) ListIDs(ctx context.Context, limit int, cursor string) ([]string, string, error) {
q := s.query ids, next, err := s.c.ListIDs(ctx, "in:inbox", limit, cursor)
if q == "" {
q = "in:inbox"
}
ids, next, err := s.c.ListIDs(ctx, q, limit, cursor)
return ids, next, err return ids, next, err
} }
@@ -220,7 +207,7 @@ func Run(ctx context.Context, cfg SyncConfig) (*SyncStats, error) {
if err != nil { if err != nil {
return nil, fmt.Errorf("gmail init: %w", err) return nil, fmt.Errorf("gmail init: %w", err)
} }
sources = append(sources, &gmailSource{c: gm, query: cfg.Query}) sources = append(sources, &gmailSource{c: gm})
} }
if len(sources) == 0 { if len(sources) == 0 {
return nil, errors.New("sync: no source configured (need OO, Gmail, or both)") return nil, errors.New("sync: no source configured (need OO, Gmail, or both)")
-65
View File
@@ -221,71 +221,6 @@ func TestCollectParts(t *testing.T) {
} }
} }
type fakeGmailAPI struct {
lastQ string
lastLimit int
ids []string
}
func (f *fakeGmailAPI) ListIDs(_ context.Context, q string, maxIDs int, _ string) ([]string, string, error) {
f.lastQ = q
f.lastLimit = maxIDs
return f.ids, "", nil
}
func (f *fakeGmailAPI) GetMessage(context.Context, string) (*Message, error) {
return nil, errors.New("unused")
}
func (f *fakeGmailAPI) DownloadAttachment(context.Context, string, string) ([]byte, error) {
return nil, errors.New("unused")
}
func TestGmailSourcePassesQueryToListIDs(t *testing.T) {
fake := &fakeGmailAPI{ids: []string{"m1"}}
src := &gmailSource{c: fake, query: "from:alice@example.com"}
ids, _, err := src.ListIDs(context.Background(), 10, "")
if err != nil {
t.Fatal(err)
}
if fake.lastQ != "from:alice@example.com" {
t.Fatalf("ListIDs q=%q, want from:alice@example.com", fake.lastQ)
}
if fake.lastLimit != 10 {
t.Fatalf("ListIDs limit=%d, want 10", fake.lastLimit)
}
if len(ids) != 1 || ids[0] != "m1" {
t.Fatalf("ids=%v", ids)
}
}
func TestGmailSourceEmptyQueryDefaultsToInbox(t *testing.T) {
fake := &fakeGmailAPI{}
src := &gmailSource{c: fake, query: ""}
if _, _, err := src.ListIDs(context.Background(), 5, ""); err != nil {
t.Fatal(err)
}
if fake.lastQ != "in:inbox" {
t.Fatalf("empty query q=%q, want in:inbox", fake.lastQ)
}
}
func TestParseCLIGmailQuery(t *testing.T) {
cli, code, err := ParseCLI([]string{
"--source", "gmail",
"--query", "from:letrado@example.com",
"--out", t.TempDir(),
"--dry-run",
})
if err != nil || code != 0 {
t.Fatalf("ParseCLI: code=%d err=%v", code, err)
}
if cli.Sync.Query != "from:letrado@example.com" {
t.Fatalf("query=%q", cli.Sync.Query)
}
if cli.Sync.Gmail == nil {
t.Fatal("gmail source not configured")
}
}
func b64(s string) string { func b64(s string) string {
return base64.URLEncoding.EncodeToString([]byte(s)) return base64.URLEncoding.EncodeToString([]byte(s))
} }
-1
View File
@@ -6,6 +6,5 @@ Brain/ops/eSlider stack. Facts need proof or they are
- [PLAN.md](../PLAN.md) — decisions, execution order, open questions (v2) - [PLAN.md](../PLAN.md) — decisions, execution order, open questions (v2)
- [design](design.md) — schema, deduction model, sources - [design](design.md) — schema, deduction model, sources
- [Gitea issues](https://git.produktor.io/eSlider/2dph/issues) — work board (origin)
Published docs live here and mirror the project state. Published docs live here and mirror the project state.
-33
View File
@@ -1,33 +0,0 @@
# CRM association proof (oo CLI ↔ corpus)
Proven with `oo` (eslider/go-onlyoffice) against the OnlyOffice portal
(`office.produktor.io`). Portal CRM is the SSOT for company ↔ person ↔
project associations; the corpus SoT (`eslider/cv/projects/knowledge-mesh-seed.yaml`)
is the second, independent source. Facts that can be backed by both are
written to the brain under `root=facts` by `bin/facts/crm`.
## What was verified
- Logical counts (portal MySQL): 1300 contacts = 897 persons + 404 companies,
198 projects, 998 deals, 939 project↔contact links.
- Every client company linked to a project has ≥1 person underneath.
- Every person `company_id` resolves to an existing company.
- Corpus org list (9) maps 1:1 onto CRM companies:
ProProdukt SL / produktor.io, Dyvenia, Immowelt AG, WhereGroup,
Keynote SIGOS, D2S/SYSTEMS, GRID, Pack und Cup, Markets Platform.
- 78 person↔company association facts written to the brain
(`how=crm-crosscheck`, `type=association`). Recall@5 in `bin/kb/eval` = 1.0.
## Mistakes found
| # | Mistake | Fix |
|---|---------|-----|
| 1 | Duplicate legal entity `GoldenRatio.Exchange` (contact 759) vs `Golden Ratio Exchange` (763); 3 deals (211, 287, 559) were linked to 759 | `oo contacts merge 759 763` — 763 kept, 759 removed, deal links re-pointed to 763 |
| 2 | `env/`-wide: OnlyOffice creds file used wrong UX (user `eslider`, password with `$2` suffix) making `oo` auth fail | `.env` fixed to `eslider@gmail.com` + clean password; `.env` stays gitignored |
## Gates after fix
- `uv run python -m unittest discover -s bin/tools -t .` → 26 tests OK
- `bin/facts/audit self` + `bin/facts/audit db` → ok
- `bin/kb/eval` → recall@5 = 1.0
- `go test ./...` (bin/server + bin/watch) → ok