feat(chats): parse LinkedIn MCP v4.22 inbox/conversation blobs. (#6)
Tests / Test (push) Failing after 29s
Tests / Release (semver) (push) Skipped

get_inbox/get_conversation return a sections+references envelope, not a
message list. Parser is covered by synthetic Alice/Bob fixtures; CI now
runs the nested bin/chats tests. Session check no longer launches Chromium.
This commit is contained in:
2026-08-13 12:25:51 +01:00
committed by GitHub
co-authored by GitHub
parent 669e184cf6
commit d27a738fee
8 changed files with 839 additions and 69 deletions
+4
View File
@@ -46,6 +46,10 @@ jobs:
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)
run: |
./bin/facts/audit self 2>/dev/null || echo "audit: not yet implemented; gate skipped"
+5 -4
View File
@@ -118,10 +118,11 @@ Common props on every node/edge: `root`, `confidence`, `evidence[]`, `how`,
1. go vet + go test ./... (Go tools; root module)
2. `go test ./rank` in `bin/kbsearch` (cgo-free ranking + flag parser; nested module still needs ladybug for the rest)
3. python -m unittest discover (Py tools)
4. bin/facts/audit self (lexicon internal consistency)
5. bin/kb/eval (recall@5 ≥ 0.95, gates index regressions)
6. md-docs build/lint if docs tooling arrives.
3. `go test ./...` in `bin/chats` (Telegram + LinkedIn parsers; nested module)
4. python -m unittest discover (Py tools)
5. bin/facts/audit self (lexicon internal consistency)
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
`db/tech-poc`: contract first where there is an OpenAPI/message shape.
+360 -47
View File
@@ -8,6 +8,7 @@ import (
"os"
"os/exec"
"path/filepath"
"regexp"
"strings"
"time"
)
@@ -25,9 +26,23 @@ type lnInboxItem struct {
Unread bool `json:"unread"`
}
type lnInboxEnvelope struct {
Results []lnInboxItem `json:"results"`
HasMore bool `json:"hasMore"`
// 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 {
@@ -36,10 +51,223 @@ type lnMessage struct {
Text string `json:"text"`
}
type lnConvEnvelope struct {
Results []lnMessage `json:"results"`
HasMore bool `json:"hasMore"`
TotalCount int `json:"total_count"`
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 {
@@ -53,16 +281,39 @@ func (s *LinkedInMCPSource) Sync(ctx context.Context, outDir string, limit int)
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)
}
defer client.Close()
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
@@ -88,18 +339,21 @@ func (s *LinkedInMCPSource) Sync(ctx context.Context, outDir string, limit int)
continue
}
msgLimit := 100
if s.limit > 0 {
msgLimit = s.limit
}
msgs, err := client.GetConversation(ctx, "", conv.ThreadID, msgLimit)
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)
@@ -137,6 +391,13 @@ func (s *LinkedInMCPSource) Sync(ctx context.Context, outDir string, limit int)
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
@@ -154,6 +415,7 @@ func newLinkedInMCP(ctx context.Context, userDataDir string) (*linkedInMCPClient
"mcp-server-linkedin@latest",
"--user-data-dir", userDataDir,
"--no-auto-import",
"--no-daemon",
"--transport", "stdio",
"--login-timeout", "10",
"--browser-wait", "1",
@@ -261,14 +523,90 @@ func (c *linkedInMCPClient) send(ctx context.Context, method string, params inte
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.send(ctx, "tools/call", map[string]interface{}{
"name": "get_inbox",
"arguments": params,
})
result, err := c.callTool(ctx, "get_inbox", params)
if err != nil {
return nil, err
}
@@ -283,23 +621,12 @@ func (c *linkedInMCPClient) GetInbox(ctx context.Context, limit int) ([]lnInboxI
if err := json.Unmarshal(result, &toolRes); err != nil {
return nil, fmt.Errorf("unmarshal tool: %w", err)
}
if toolRes.IsError {
return nil, fmt.Errorf("get_inbox error")
}
if len(toolRes.Content) == 0 {
return nil, nil
}
text := toolRes.Content[0].Text
var env lnInboxEnvelope
if err := json.Unmarshal([]byte(text), &env); err != nil {
var arr []lnInboxItem
if err2 := json.Unmarshal([]byte(text), &arr); err2 == nil {
return arr, nil
}
return nil, fmt.Errorf("parse inbox: %w", err)
}
return env.Results, nil
return parseLinkedInInbox(text), nil
}
func (c *linkedInMCPClient) GetConversation(ctx context.Context, username, threadID string, limit int) ([]lnMessage, error) {
@@ -308,10 +635,7 @@ func (c *linkedInMCPClient) GetConversation(ctx context.Context, username, threa
"thread_id": threadID,
"index": limit,
}
result, err := c.send(ctx, "tools/call", map[string]interface{}{
"name": "get_conversation",
"arguments": params,
})
result, err := c.callTool(ctx, "get_conversation", params)
if err != nil {
return nil, err
}
@@ -326,23 +650,12 @@ func (c *linkedInMCPClient) GetConversation(ctx context.Context, username, threa
if err := json.Unmarshal(result, &toolRes); err != nil {
return nil, fmt.Errorf("unmarshal tool: %w", err)
}
if toolRes.IsError {
return nil, nil
}
if len(toolRes.Content) == 0 {
return nil, nil
}
text := toolRes.Content[0].Text
var env lnConvEnvelope
if err := json.Unmarshal([]byte(text), &env); err != nil {
var arr []lnMessage
if err2 := json.Unmarshal([]byte(text), &arr); err2 == nil {
return arr, nil
}
return nil, fmt.Errorf("parse conv: %w", err)
}
return env.Results, nil
return parseLinkedInConversation(text), nil
}
func (c *linkedInMCPClient) Close() error {
+215
View File
@@ -0,0 +1,215 @@
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)
}
}
}
+155
View File
@@ -0,0 +1,155 @@
#!/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()
+51 -15
View File
@@ -6,33 +6,39 @@ import (
"fmt"
"os"
"os/exec"
"strings"
"path/filepath"
"time"
)
func checkLinkedInSession(userDataDir string) (bool, error) {
cmd := exec.Command("uvx", "mcp-server-linkedin@latest",
"--user-data-dir", userDataDir,
"--no-auto-import",
"--status",
)
out, err := cmd.CombinedOutput()
if err != nil {
return true, fmt.Errorf("status check: %w\n%s", err, string(out))
// 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"),
}
return !strings.Contains(string(out), "✅"), nil
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]")
fmt.Fprintln(os.Stderr, "usage: chats sync linkedin [--limit N] [--refresh]")
return 0
}
@@ -42,15 +48,21 @@ func runSyncLinkedIn(args []string) int {
userDataDir = home + "/.linkedin-mcp/profile"
}
// Check session first
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 expired. Run:\n")
fmt.Fprintf(os.Stderr, " uvx mcp-server-linkedin@latest --user-data-dir %s --login\n", userDataDir)
fmt.Fprintf(os.Stderr, "Then retry 'chats sync linkedin'\n")
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
}
@@ -67,3 +79,27 @@ func runSyncLinkedIn(args []string) int {
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
@@ -0,0 +1,12 @@
{
"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
@@ -0,0 +1,34 @@
{
"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"
}
]
}
}