mirror of
https://github.com/lukaszraczylo/claude-mnemonic.git
synced 2026-10-12 04:55:18 +00:00
f2886bbedb
After 30 minutes without a message the idle monitor closed stdin and expected the read loop to end. On a real, blocking stdin a read already in progress is not woken by closing the file, so the server lived on, the client's next message completed that read and the one after failed with 'read /dev/stdin: file already closed', which ended the connection mid-call. The idle shutdown is now off by default (a stdio server lives as long as its client; orphans are still caught by the parent check). CLAUDE_MNEMONIC_MCP_IDLE_TIMEOUT (a Go duration) turns it on, and then it cancels the loop's context so Run returns nil instead of closing stdin. Co-authored-by: Claude Sonnet 5.5 <[email protected]>
1645 lines
53 KiB
Go
1645 lines
53 KiB
Go
// Package mcp provides the MCP (Model Context Protocol) server for claude-mnemonic.
|
|
package mcp
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"strconv"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
// Server is the MCP server that proxies tool calls to the worker HTTP API.
|
|
// Field order optimized for memory alignment (fieldalignment).
|
|
type Server struct {
|
|
stdin io.Reader
|
|
stdout io.Writer
|
|
client *http.Client
|
|
workerURL string
|
|
project string
|
|
version string
|
|
writeMu sync.Mutex
|
|
lastActivity atomic.Int64
|
|
// idleTimeout ends the server after this long without a message; 0 (the default) never does. A stdio server lives
|
|
// as long as its client: ending it while the client is still connected makes the client's next call fail.
|
|
idleTimeout time.Duration
|
|
// monitorInterval is how often the parent and idle checks run (30s when 0).
|
|
monitorInterval time.Duration
|
|
}
|
|
|
|
// NewServer creates a new MCP server that proxies to the worker HTTP API.
|
|
func NewServer(client *http.Client, workerURL, project, version string) *Server {
|
|
return &Server{
|
|
client: client,
|
|
workerURL: workerURL,
|
|
project: project,
|
|
version: version,
|
|
stdin: os.Stdin,
|
|
stdout: os.Stdout,
|
|
|
|
idleTimeout: idleTimeoutFromEnv(),
|
|
}
|
|
}
|
|
|
|
// IdleTimeoutEnv names the environment variable that turns the idle shutdown on, as a Go duration such as "2h".
|
|
// It is off by default.
|
|
const IdleTimeoutEnv = "CLAUDE_MNEMONIC_MCP_IDLE_TIMEOUT"
|
|
|
|
// idleTimeoutFromEnv reads IdleTimeoutEnv; an unset, invalid or non-positive value means no idle shutdown.
|
|
func idleTimeoutFromEnv() time.Duration {
|
|
d, err := time.ParseDuration(os.Getenv(IdleTimeoutEnv))
|
|
if err != nil || d <= 0 {
|
|
return 0
|
|
}
|
|
return d
|
|
}
|
|
|
|
// Request represents a JSON-RPC request.
|
|
type Request struct {
|
|
JSONRPC string `json:"jsonrpc"`
|
|
ID any `json:"id"`
|
|
Method string `json:"method"`
|
|
Params json.RawMessage `json:"params,omitempty"`
|
|
}
|
|
|
|
// Response represents a JSON-RPC response.
|
|
type Response struct {
|
|
ID any `json:"id"`
|
|
Result any `json:"result,omitempty"`
|
|
Error *Error `json:"error,omitempty"`
|
|
JSONRPC string `json:"jsonrpc"`
|
|
}
|
|
|
|
// Error represents a JSON-RPC error.
|
|
type Error struct {
|
|
Data any `json:"data,omitempty"`
|
|
Message string `json:"message"`
|
|
Code int `json:"code"`
|
|
}
|
|
|
|
// ToolCallParams represents parameters for tools/call method.
|
|
type ToolCallParams struct {
|
|
Name string `json:"name"`
|
|
Arguments json.RawMessage `json:"arguments"`
|
|
}
|
|
|
|
// Tool represents an MCP tool definition.
|
|
type Tool struct {
|
|
InputSchema map[string]any `json:"inputSchema"`
|
|
Name string `json:"name"`
|
|
Description string `json:"description"`
|
|
}
|
|
|
|
// Run starts the MCP server loop.
|
|
func (s *Server) Run(ctx context.Context) error {
|
|
// An idle shutdown cancels this context. Closing stdin does not do it: on a real (blocking) stdin the read stays
|
|
// blocked, the server lived on with a closed stdin, and the client's next message woke the read with "file already
|
|
// closed", which killed the connection mid-call.
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
var idleStopped atomic.Bool
|
|
|
|
scanner := bufio.NewScanner(s.stdin)
|
|
buf := make([]byte, 0, 64*1024)
|
|
scanner.Buffer(buf, 1024*1024) // 1MB max message size
|
|
|
|
lines := make(chan string)
|
|
scanErr := make(chan error, 1)
|
|
|
|
// Track last activity for idle timeout
|
|
s.lastActivity.Store(time.Now().Unix())
|
|
|
|
go func() {
|
|
defer close(lines)
|
|
for scanner.Scan() {
|
|
lines <- scanner.Text()
|
|
}
|
|
scanErr <- scanner.Err()
|
|
}()
|
|
|
|
// Monitor parent process liveness and, when one is set, the idle timeout.
|
|
// If the parent dies (ppid changes) or no messages arrive for idleTimeout, shut down.
|
|
parentPID := os.Getppid()
|
|
go func() {
|
|
interval := s.monitorInterval
|
|
if interval <= 0 {
|
|
interval = 30 * time.Second
|
|
}
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
if os.Getppid() != parentPID {
|
|
log.Info().Msg("Parent process died, shutting down MCP server")
|
|
if closer, ok := s.stdin.(io.Closer); ok {
|
|
_ = closer.Close()
|
|
}
|
|
return
|
|
}
|
|
if s.idleTimeout > 0 && time.Since(time.Unix(s.lastActivity.Load(), 0)) > s.idleTimeout {
|
|
log.Info().Dur("idle", s.idleTimeout).Msg("MCP server idle timeout, shutting down")
|
|
idleStopped.Store(true)
|
|
cancel()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
|
|
// Semaphore limits concurrent request goroutines.
|
|
const maxConcurrent = 10
|
|
sem := make(chan struct{}, maxConcurrent)
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
// Drain in-flight requests before returning.
|
|
wg.Wait()
|
|
if idleStopped.Load() {
|
|
return nil // an idle shutdown the operator asked for is a clean exit
|
|
}
|
|
return ctx.Err()
|
|
case line, ok := <-lines:
|
|
if !ok {
|
|
// Scanner finished — drain in-flight requests, then check for errors.
|
|
wg.Wait()
|
|
err := <-scanErr
|
|
if err != nil {
|
|
if errors.Is(err, bufio.ErrTooLong) {
|
|
log.Error().Msg("MCP message exceeded 1MB buffer limit")
|
|
}
|
|
return fmt.Errorf("scanner error: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
s.lastActivity.Store(time.Now().Unix())
|
|
|
|
if line == "" {
|
|
continue
|
|
}
|
|
|
|
var req Request
|
|
if err := json.Unmarshal([]byte(line), &req); err != nil {
|
|
// Parse errors are cheap — send inline, no goroutine needed.
|
|
if werr := s.sendError(nil, -32700, "Parse error", err.Error()); werr != nil {
|
|
return fmt.Errorf("write error: %w", werr)
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Dispatch request to its own goroutine.
|
|
// Semaphore is acquired inside the goroutine so the main
|
|
// loop never blocks on a full semaphore (Fix #1).
|
|
wg.Add(1)
|
|
go func(r Request) {
|
|
defer wg.Done()
|
|
// Acquire semaphore inside goroutine, not blocking main loop
|
|
select {
|
|
case sem <- struct{}{}:
|
|
defer func() { <-sem }()
|
|
case <-ctx.Done():
|
|
// Server shutting down — send error so caller isn't left waiting
|
|
if r.ID != nil {
|
|
_ = s.sendError(r.ID, -32000, "Server shutting down", nil)
|
|
}
|
|
return
|
|
}
|
|
|
|
resp := s.handleRequest(ctx, &r)
|
|
if resp != nil {
|
|
if werr := s.sendResponse(resp); werr != nil {
|
|
log.Error().Err(werr).Msg("Failed to send response")
|
|
}
|
|
}
|
|
}(req)
|
|
}
|
|
}
|
|
}
|
|
|
|
// handleRequest dispatches the request to the appropriate handler.
|
|
func (s *Server) handleRequest(ctx context.Context, req *Request) *Response {
|
|
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
|
defer cancel()
|
|
|
|
switch req.Method {
|
|
case "initialize":
|
|
return s.handleInitialize(req)
|
|
case "tools/list":
|
|
return s.handleToolsList(req)
|
|
case "tools/call":
|
|
return s.handleToolsCall(ctx, req)
|
|
case "notifications/initialized", "notifications/cancelled":
|
|
return nil // Notifications don't get responses
|
|
default:
|
|
if req.ID == nil {
|
|
// Notifications must not receive responses per JSON-RPC 2.0
|
|
return nil
|
|
}
|
|
return &Response{
|
|
JSONRPC: "2.0",
|
|
ID: req.ID,
|
|
Error: &Error{
|
|
Code: -32601,
|
|
Message: "Method not found",
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
// handleInitialize handles the initialize request.
|
|
func (s *Server) handleInitialize(req *Request) *Response {
|
|
return &Response{
|
|
JSONRPC: "2.0",
|
|
ID: req.ID,
|
|
Result: map[string]any{
|
|
"protocolVersion": "2024-11-05",
|
|
"capabilities": map[string]any{
|
|
"tools": map[string]any{},
|
|
},
|
|
"serverInfo": map[string]any{
|
|
"name": "claude-mnemonic",
|
|
"version": s.version,
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
// handleToolsList returns the list of available tools.
|
|
func (s *Server) handleToolsList(req *Request) *Response {
|
|
tools := []Tool{
|
|
{
|
|
Name: "search",
|
|
Description: "Unified search across all memory types (observations, sessions, and user prompts) using vector-first semantic search (sqlite-vec).",
|
|
InputSchema: map[string]any{
|
|
"type": "object",
|
|
"properties": map[string]any{
|
|
"query": map[string]any{"type": "string", "description": "Natural language search query for semantic ranking"},
|
|
"type": map[string]any{"type": "string", "enum": []string{"observations", "sessions", "prompts"}, "description": "Filter by document type"},
|
|
"project": map[string]any{"type": "string", "description": "Filter by project name"},
|
|
"obs_type": map[string]any{"type": "string", "description": "Filter observations by type"},
|
|
"concepts": map[string]any{"type": "string", "description": "Filter by concept tags"},
|
|
"files": map[string]any{"type": "string", "description": "Filter by file paths"},
|
|
"dateStart": map[string]any{"type": []string{"string", "number"}, "description": "Start date for filtering"},
|
|
"dateEnd": map[string]any{"type": []string{"string", "number"}, "description": "End date for filtering"},
|
|
"orderBy": map[string]any{"type": "string", "enum": []string{"relevance", "date_desc", "date_asc"}, "default": "date_desc"},
|
|
"limit": map[string]any{"type": "number", "default": 20, "minimum": 1, "maximum": 100},
|
|
"offset": map[string]any{"type": "number", "default": 0, "minimum": 0},
|
|
"format": map[string]any{"type": "string", "enum": []string{"index", "full"}, "default": "index"},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Name: "timeline",
|
|
Description: "Fetch timeline of observations around a specific point in time.",
|
|
InputSchema: map[string]any{
|
|
"type": "object",
|
|
"properties": map[string]any{
|
|
"anchor_id": map[string]any{"type": "number", "description": "Observation ID to use as anchor"},
|
|
"query": map[string]any{"type": "string", "description": "Natural language query to find anchor observation"},
|
|
"before": map[string]any{"type": "number", "default": 10, "minimum": 0, "maximum": 100},
|
|
"after": map[string]any{"type": "number", "default": 10, "minimum": 0, "maximum": 100},
|
|
"project": map[string]any{"type": "string"},
|
|
"concepts": map[string]any{"type": "string"},
|
|
"files": map[string]any{"type": "string"},
|
|
"obs_type": map[string]any{"type": "string"},
|
|
"format": map[string]any{"type": "string", "enum": []string{"index", "full"}, "default": "index"},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Name: "observation",
|
|
Description: "Manage individual observations. Set 'action' to choose the operation: get, edit, delete, supersede, boost, merge, related, similar, quality, relationships, scoring, tag, by_tag, batch_tag.",
|
|
InputSchema: map[string]any{
|
|
"type": "object",
|
|
"required": []string{"action"},
|
|
"properties": map[string]any{
|
|
"action": map[string]any{"type": "string", "enum": []string{"get", "edit", "delete", "supersede", "boost", "merge", "related", "similar", "quality", "relationships", "scoring", "tag", "by_tag", "batch_tag"}, "description": "Operation to perform"},
|
|
"id": map[string]any{"type": "number", "description": "Observation ID (get, edit, quality, relationships, scoring, related)"},
|
|
"ids": map[string]any{"type": "array", "items": map[string]any{"type": "number"}, "description": "Observation IDs (delete, supersede, boost)"},
|
|
"title": map[string]any{"type": "string", "description": "edit: new title"},
|
|
"subtitle": map[string]any{"type": "string", "description": "edit: new subtitle"},
|
|
"narrative": map[string]any{"type": "string", "description": "edit: new narrative"},
|
|
"facts": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "edit: new facts"},
|
|
"concepts": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "edit: new concept tags"},
|
|
"files_read": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "edit: new files read"},
|
|
"files_modified": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "edit: new files modified"},
|
|
"scope": map[string]any{"type": "string", "enum": []string{"project", "global"}, "description": "edit: new scope"},
|
|
"query": map[string]any{"type": "string", "description": "similar: text to find similar observations for"},
|
|
"project": map[string]any{"type": "string", "description": "Filter by project (similar, by_tag, batch_tag)"},
|
|
"min_confidence": map[string]any{"type": "number", "minimum": 0.0, "maximum": 1.0, "description": "related: minimum confidence (default 0.5)"},
|
|
"min_similarity": map[string]any{"type": "number", "minimum": 0.0, "maximum": 1.0, "description": "similar: minimum similarity (default 0.7)"},
|
|
"limit": map[string]any{"type": "number", "description": "Max results (related, similar, by_tag, batch_tag)"},
|
|
"delete_vectors": map[string]any{"type": "boolean", "description": "delete: also delete vectors (default true)"},
|
|
"boost": map[string]any{"type": "number", "minimum": -1.0, "maximum": 1.0, "description": "boost: amount; merge: target boost"},
|
|
"source_id": map[string]any{"type": "number", "description": "merge: source observation ID (superseded)"},
|
|
"target_id": map[string]any{"type": "number", "description": "merge: target observation ID (kept)"},
|
|
"max_depth": map[string]any{"type": "number", "minimum": 1, "maximum": 5, "description": "relationships: traversal hops (default 2)"},
|
|
"tag": map[string]any{"type": "string", "description": "by_tag: tag/concept to search for"},
|
|
"tags": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "tag/batch_tag: tags to apply"},
|
|
"mode": map[string]any{"type": "string", "enum": []string{"add", "remove", "set"}, "description": "tag: add/remove/set (default add)"},
|
|
"pattern": map[string]any{"type": "string", "description": "batch_tag: search pattern to match"},
|
|
"dry_run": map[string]any{"type": "boolean", "description": "batch_tag: preview only (default true)"},
|
|
"max_matches": map[string]any{"type": "number", "description": "batch_tag: max observations to tag"},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Name: "memory_admin",
|
|
Description: "Memory system administration and analytics. Set 'action': stats, health, maintenance_stats, run_maintenance, importance, search_patterns, explain_ranking, temporal_trends, data_quality, export, suggest_consolidations, patterns.",
|
|
InputSchema: map[string]any{
|
|
"type": "object",
|
|
"required": []string{"action"},
|
|
"properties": map[string]any{
|
|
"action": map[string]any{"type": "string", "enum": []string{"stats", "health", "maintenance_stats", "run_maintenance", "importance", "search_patterns", "explain_ranking", "temporal_trends", "data_quality", "export", "suggest_consolidations", "patterns"}, "description": "Operation to perform"},
|
|
"project": map[string]any{"type": "string", "description": "Filter by project (importance, temporal_trends, data_quality, export, suggest_consolidations, patterns, explain_ranking)"},
|
|
"query": map[string]any{"type": "string", "description": "explain_ranking: query to analyze; patterns: search by name/description"},
|
|
"top_n": map[string]any{"type": "number", "description": "Top results (explain_ranking, search_patterns)"},
|
|
"days": map[string]any{"type": "number", "description": "Days to analyze (search_patterns, temporal_trends)"},
|
|
"group_by": map[string]any{"type": "string", "enum": []string{"day", "week", "hour_of_day"}, "description": "temporal_trends: grouping (default day)"},
|
|
"limit": map[string]any{"type": "number", "description": "Max results (importance, data_quality, export, suggest_consolidations, patterns)"},
|
|
"min_similarity": map[string]any{"type": "number", "minimum": 0.5, "maximum": 1.0, "description": "suggest_consolidations: grouping threshold (default 0.8)"},
|
|
"format": map[string]any{"type": "string", "enum": []string{"json", "jsonl", "markdown"}, "description": "export: output format (default json)"},
|
|
"date_start": map[string]any{"type": "number", "description": "export: filter start (epoch ms)"},
|
|
"date_end": map[string]any{"type": "number", "description": "export: filter end (epoch ms)"},
|
|
"obs_type": map[string]any{"type": "string", "description": "export: filter by observation type"},
|
|
"type": map[string]any{"type": "string", "enum": []string{"workflow", "preference", "best_practice", "anti_pattern", "tooling"}, "description": "patterns: filter by pattern type"},
|
|
"include_top_scored": map[string]any{"type": "boolean", "description": "importance: include top-scoring (default true)"},
|
|
"include_most_retrieved": map[string]any{"type": "boolean", "description": "importance: include most-retrieved (default true)"},
|
|
"include_concept_weights": map[string]any{"type": "boolean", "description": "importance: include concept weights (default true)"},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
return &Response{
|
|
JSONRPC: "2.0",
|
|
ID: req.ID,
|
|
Result: map[string]any{
|
|
"tools": tools,
|
|
},
|
|
}
|
|
}
|
|
|
|
// handleToolsCall handles tool invocations.
|
|
func (s *Server) handleToolsCall(ctx context.Context, req *Request) *Response {
|
|
var params ToolCallParams
|
|
if err := json.Unmarshal(req.Params, ¶ms); err != nil {
|
|
return &Response{
|
|
JSONRPC: "2.0",
|
|
ID: req.ID,
|
|
Error: &Error{
|
|
Code: -32602,
|
|
Message: "Invalid params",
|
|
Data: err.Error(),
|
|
},
|
|
}
|
|
}
|
|
|
|
result, err := s.callTool(ctx, params.Name, params.Arguments)
|
|
if err != nil {
|
|
// MCP spec: tool failures use Result with isError, not top-level Error
|
|
return &Response{
|
|
JSONRPC: "2.0",
|
|
ID: req.ID,
|
|
Result: map[string]any{
|
|
"content": []map[string]any{
|
|
{"type": "text", "text": "Error: " + err.Error()},
|
|
},
|
|
"isError": true,
|
|
},
|
|
}
|
|
}
|
|
|
|
return &Response{
|
|
JSONRPC: "2.0",
|
|
ID: req.ID,
|
|
Result: map[string]any{
|
|
"content": []map[string]any{
|
|
{
|
|
"type": "text",
|
|
"text": result,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
// searchArgs holds common search parameters used by many tools.
|
|
type searchArgs struct {
|
|
Query string `json:"query"`
|
|
Project string `json:"project"`
|
|
Type string `json:"type"`
|
|
ObsType string `json:"obs_type"`
|
|
Concepts string `json:"concepts"`
|
|
Files string `json:"files"`
|
|
DateStart any `json:"dateStart"`
|
|
DateEnd any `json:"dateEnd"`
|
|
OrderBy string `json:"orderBy"`
|
|
Format string `json:"format"`
|
|
Limit int `json:"limit"`
|
|
Offset int `json:"offset"`
|
|
}
|
|
|
|
// observationActions maps the 'observation' tool's action values to the internal tool name that implements them.
|
|
var observationActions = map[string]string{
|
|
"get": "get_observation",
|
|
"edit": "edit_observation",
|
|
"delete": "bulk_delete_observations",
|
|
"supersede": "bulk_mark_superseded",
|
|
"boost": "bulk_boost_observations",
|
|
"merge": "merge_observations",
|
|
"related": "find_related_observations",
|
|
"similar": "find_similar_observations",
|
|
"quality": "get_observation_quality",
|
|
"relationships": "get_observation_relationships",
|
|
"scoring": "get_observation_scoring_breakdown",
|
|
"tag": "tag_observation",
|
|
"by_tag": "get_observations_by_tag",
|
|
"batch_tag": "batch_tag_by_pattern",
|
|
}
|
|
|
|
// adminActions maps the 'memory_admin' tool's action values to the internal tool name that implements them.
|
|
var adminActions = map[string]string{
|
|
"stats": "get_memory_stats",
|
|
"health": "check_system_health",
|
|
"maintenance_stats": "get_maintenance_stats",
|
|
"run_maintenance": "trigger_maintenance",
|
|
"importance": "analyze_observation_importance",
|
|
"search_patterns": "analyze_search_patterns",
|
|
"explain_ranking": "explain_search_ranking",
|
|
"temporal_trends": "get_temporal_trends",
|
|
"data_quality": "get_data_quality_report",
|
|
"export": "export_observations",
|
|
"suggest_consolidations": "suggest_consolidations",
|
|
"patterns": "get_patterns",
|
|
}
|
|
|
|
// dispatchAction routes a multiplexed tool call to its underlying implementation by reading the "action" field.
|
|
func (s *Server) dispatchAction(ctx context.Context, tool string, actions map[string]string, args json.RawMessage) (string, error) {
|
|
var a struct {
|
|
Action string `json:"action"`
|
|
}
|
|
if err := json.Unmarshal(args, &a); err != nil {
|
|
return "", fmt.Errorf("%s: invalid arguments: %w", tool, err)
|
|
}
|
|
target, ok := actions[a.Action]
|
|
if !ok {
|
|
return "", fmt.Errorf("%s: unknown action %q", tool, a.Action)
|
|
}
|
|
return s.callTool(ctx, target, args)
|
|
}
|
|
|
|
// callTool dispatches to the appropriate tool handler by proxying to the worker HTTP API.
|
|
func (s *Server) callTool(ctx context.Context, name string, args json.RawMessage) (string, error) {
|
|
// Parse common search params used by many tools
|
|
var sa searchArgs
|
|
// Best-effort parse; individual handlers validate as needed
|
|
_ = json.Unmarshal(args, &sa)
|
|
|
|
// Default project from server config
|
|
if sa.Project == "" {
|
|
sa.Project = s.project
|
|
}
|
|
|
|
switch name {
|
|
// --- Search-based tools: proxy to GET /api/context/search ---
|
|
case "search":
|
|
return s.handleSearchProxy(ctx, sa)
|
|
case "observation":
|
|
return s.dispatchAction(ctx, "observation", observationActions, args)
|
|
case "memory_admin":
|
|
return s.dispatchAction(ctx, "memory_admin", adminActions, args)
|
|
case "decisions":
|
|
sa.ObsType = "decision"
|
|
return s.handleSearchProxy(ctx, sa)
|
|
case "changes":
|
|
sa.ObsType = "code_change"
|
|
return s.handleSearchProxy(ctx, sa)
|
|
case "how_it_works":
|
|
sa.ObsType = "architecture"
|
|
return s.handleSearchProxy(ctx, sa)
|
|
case "find_by_concept":
|
|
return s.handleSearchProxy(ctx, sa)
|
|
case "find_by_file":
|
|
return s.handleSearchProxy(ctx, sa)
|
|
case "find_by_type":
|
|
return s.handleSearchProxy(ctx, sa)
|
|
case "get_recent_context":
|
|
return s.handleSearchProxy(ctx, sa)
|
|
case "timeline", "get_context_timeline":
|
|
return s.handleTimelineProxy(ctx, args)
|
|
case "get_timeline_by_query":
|
|
return s.handleTimelineProxy(ctx, args)
|
|
|
|
// --- Observation endpoints ---
|
|
case "get_observation":
|
|
return s.handleGetObservationProxy(ctx, args)
|
|
case "edit_observation":
|
|
return s.handleEditObservationProxy(ctx, args)
|
|
case "find_related_observations":
|
|
return s.handleFindRelatedProxy(ctx, args)
|
|
case "find_similar_observations":
|
|
return s.handleFindSimilarProxy(ctx, args)
|
|
case "get_observation_quality":
|
|
return s.handleGetObservationQualityProxy(ctx, args)
|
|
case "get_observation_relationships":
|
|
return s.handleGetRelationshipsProxy(ctx, args)
|
|
case "get_observation_scoring_breakdown":
|
|
return s.handleGetScoringBreakdownProxy(ctx, args)
|
|
case "tag_observation":
|
|
return s.handleTagObservationProxy(ctx, args)
|
|
case "get_observations_by_tag":
|
|
return s.handleGetObservationsByTagProxy(ctx, args)
|
|
|
|
// --- Bulk operations ---
|
|
case "bulk_delete_observations":
|
|
return s.handleBulkStatusProxy(ctx, args, "delete")
|
|
case "bulk_mark_superseded":
|
|
return s.handleBulkStatusProxy(ctx, args, "supersede")
|
|
case "bulk_boost_observations":
|
|
return s.handleBulkStatusProxy(ctx, args, "boost")
|
|
case "merge_observations":
|
|
return s.handleMergeProxy(ctx, args)
|
|
|
|
// --- Pattern endpoints ---
|
|
case "get_patterns":
|
|
return s.handleGetPatternsProxy(ctx, args)
|
|
|
|
// --- Stats and analytics endpoints ---
|
|
case "get_memory_stats":
|
|
return s.proxyGetRaw(ctx, "/api/stats", map[string]string{
|
|
"project": s.project,
|
|
})
|
|
case "check_system_health":
|
|
return s.proxyGetRaw(ctx, "/api/selfcheck", nil)
|
|
case "get_maintenance_stats":
|
|
return s.proxyGetRaw(ctx, "/api/stats", map[string]string{
|
|
"project": s.project,
|
|
})
|
|
case "trigger_maintenance":
|
|
return s.proxyPostRaw(ctx, "/api/maintenance/run", nil)
|
|
case "analyze_observation_importance":
|
|
return s.handleAnalyzeImportanceProxy(ctx, args)
|
|
case "analyze_search_patterns":
|
|
return s.proxyGetRaw(ctx, "/api/search/analytics", nil)
|
|
case "explain_search_ranking":
|
|
return s.handleExplainSearchProxy(ctx, args)
|
|
case "get_temporal_trends":
|
|
return s.handleGetTemporalTrendsProxy(ctx, args)
|
|
case "get_data_quality_report":
|
|
return s.handleGetDataQualityProxy(ctx, args)
|
|
|
|
// --- Export and batch tag ---
|
|
case "export_observations":
|
|
return s.handleExportProxy(ctx, args)
|
|
case "suggest_consolidations":
|
|
return s.handleSuggestConsolidationsProxy(ctx, args)
|
|
case "batch_tag_by_pattern":
|
|
return s.handleBatchTagProxy(ctx, args)
|
|
|
|
default:
|
|
return "", fmt.Errorf("unknown tool: %s", name)
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// HTTP PROXY HELPERS
|
|
// =============================================================================
|
|
|
|
// proxyGetRaw performs a GET request to the worker and returns the raw JSON response body.
|
|
func (s *Server) proxyGetRaw(ctx context.Context, path string, params map[string]string) (string, error) {
|
|
if s.client == nil {
|
|
return "", fmt.Errorf("worker unavailable at %s: http client not configured", s.workerURL)
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, "GET", s.workerURL+path, nil)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if len(params) > 0 {
|
|
q := req.URL.Query()
|
|
for k, v := range params {
|
|
if v != "" {
|
|
q.Set(k, v)
|
|
}
|
|
}
|
|
req.URL.RawQuery = q.Encode()
|
|
}
|
|
|
|
resp, err := s.client.Do(req)
|
|
if err != nil {
|
|
return "", fmt.Errorf("worker unavailable at %s: %w", s.workerURL, err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return "", fmt.Errorf("read worker response: %w", err)
|
|
}
|
|
|
|
if resp.StatusCode >= 400 {
|
|
return "", fmt.Errorf("worker returned %d: %s", resp.StatusCode, string(body))
|
|
}
|
|
|
|
return string(body), nil
|
|
}
|
|
|
|
// proxyPostRaw performs a POST request to the worker and returns the raw JSON response body.
|
|
func (s *Server) proxyPostRaw(ctx context.Context, path string, payload any) (string, error) {
|
|
if s.client == nil {
|
|
return "", fmt.Errorf("worker unavailable at %s: http client not configured", s.workerURL)
|
|
}
|
|
var bodyReader io.Reader
|
|
if payload != nil {
|
|
data, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
bodyReader = bytes.NewReader(data)
|
|
}
|
|
|
|
req, err := http.NewRequestWithContext(ctx, "POST", s.workerURL+path, bodyReader)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if payload != nil {
|
|
req.Header.Set("Content-Type", "application/json")
|
|
}
|
|
|
|
resp, err := s.client.Do(req)
|
|
if err != nil {
|
|
return "", fmt.Errorf("worker unavailable at %s: %w", s.workerURL, err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return "", fmt.Errorf("read worker response: %w", err)
|
|
}
|
|
|
|
if resp.StatusCode >= 400 {
|
|
return "", fmt.Errorf("worker returned %d: %s", resp.StatusCode, string(body))
|
|
}
|
|
|
|
return string(body), nil
|
|
}
|
|
|
|
// proxyPutRaw performs a PUT request to the worker and returns the raw JSON response body.
|
|
func (s *Server) proxyPutRaw(ctx context.Context, path string, payload any) (string, error) {
|
|
if s.client == nil {
|
|
return "", fmt.Errorf("worker unavailable at %s: http client not configured", s.workerURL)
|
|
}
|
|
var bodyReader io.Reader
|
|
if payload != nil {
|
|
data, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
bodyReader = bytes.NewReader(data)
|
|
}
|
|
|
|
req, err := http.NewRequestWithContext(ctx, "PUT", s.workerURL+path, bodyReader)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if payload != nil {
|
|
req.Header.Set("Content-Type", "application/json")
|
|
}
|
|
|
|
resp, err := s.client.Do(req)
|
|
if err != nil {
|
|
return "", fmt.Errorf("worker unavailable at %s: %w", s.workerURL, err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return "", fmt.Errorf("read worker response: %w", err)
|
|
}
|
|
|
|
if resp.StatusCode >= 400 {
|
|
return "", fmt.Errorf("worker returned %d: %s", resp.StatusCode, string(body))
|
|
}
|
|
|
|
return string(body), nil
|
|
}
|
|
|
|
// anyToString converts an interface{} value to its string representation for query params.
|
|
func anyToString(v any) string {
|
|
if v == nil {
|
|
return ""
|
|
}
|
|
switch val := v.(type) {
|
|
case string:
|
|
return val
|
|
case float64:
|
|
return strconv.FormatInt(int64(val), 10)
|
|
case int:
|
|
return strconv.Itoa(val)
|
|
case int64:
|
|
return strconv.FormatInt(val, 10)
|
|
default:
|
|
return fmt.Sprintf("%v", v)
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// TOOL HANDLER PROXIES
|
|
// =============================================================================
|
|
|
|
// handleSearchProxy proxies search requests to GET /api/context/search.
|
|
func (s *Server) handleSearchProxy(ctx context.Context, args searchArgs) (string, error) {
|
|
params := map[string]string{
|
|
"project": args.Project,
|
|
"query": args.Query,
|
|
}
|
|
if args.Limit > 0 {
|
|
params["limit"] = strconv.Itoa(args.Limit)
|
|
}
|
|
if args.ObsType != "" {
|
|
params["obs_type"] = args.ObsType
|
|
}
|
|
if args.Concepts != "" {
|
|
params["concepts"] = args.Concepts
|
|
}
|
|
if args.Files != "" {
|
|
params["files"] = args.Files
|
|
}
|
|
if args.DateStart != nil {
|
|
if ds := anyToString(args.DateStart); ds != "" {
|
|
params["dateStart"] = ds
|
|
}
|
|
}
|
|
if args.DateEnd != nil {
|
|
if de := anyToString(args.DateEnd); de != "" {
|
|
params["dateEnd"] = de
|
|
}
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, "/api/context/search", params)
|
|
}
|
|
|
|
// handleTimelineProxy proxies timeline requests. First searches for anchor, then fetches surrounding observations.
|
|
func (s *Server) handleTimelineProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
AnchorID int64 `json:"anchor_id"`
|
|
Query string `json:"query"`
|
|
Project string `json:"project"`
|
|
Before int `json:"before"`
|
|
After int `json:"after"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid timeline params: %w", err)
|
|
}
|
|
|
|
if params.Project == "" {
|
|
params.Project = s.project
|
|
}
|
|
if params.Before <= 0 {
|
|
params.Before = 10
|
|
}
|
|
if params.After <= 0 {
|
|
params.After = 10
|
|
}
|
|
|
|
// If query provided and no anchor, search for it first
|
|
if params.Query != "" && params.AnchorID == 0 {
|
|
searchResult, err := s.proxyGetRaw(ctx, "/api/context/search", map[string]string{
|
|
"project": params.Project,
|
|
"query": params.Query,
|
|
"limit": "1",
|
|
})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
// Extract first observation ID from search results
|
|
var searchResp struct {
|
|
Observations []struct {
|
|
ID int64 `json:"id"`
|
|
} `json:"observations"`
|
|
}
|
|
if err := json.Unmarshal([]byte(searchResult), &searchResp); err == nil && len(searchResp.Observations) > 0 {
|
|
params.AnchorID = searchResp.Observations[0].ID
|
|
}
|
|
}
|
|
|
|
if params.AnchorID == 0 {
|
|
result, _ := json.Marshal(map[string]any{"observations": []any{}, "message": "no anchor found"})
|
|
return string(result), nil
|
|
}
|
|
|
|
// Get observations around the anchor
|
|
limit := params.Before + params.After + 1
|
|
return s.proxyGetRaw(ctx, "/api/observations", map[string]string{
|
|
"project": params.Project,
|
|
"limit": strconv.Itoa(limit),
|
|
})
|
|
}
|
|
|
|
// handleGetObservationProxy proxies get observation by ID.
|
|
func (s *Server) handleGetObservationProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
ID int64 `json:"id"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.ID == 0 {
|
|
return "", fmt.Errorf("id is required")
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, fmt.Sprintf("/api/observations/%d", params.ID), nil)
|
|
}
|
|
|
|
// handleEditObservationProxy proxies observation edits.
|
|
func (s *Server) handleEditObservationProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Title *string `json:"title,omitempty"`
|
|
Subtitle *string `json:"subtitle,omitempty"`
|
|
Narrative *string `json:"narrative,omitempty"`
|
|
Scope *string `json:"scope,omitempty"`
|
|
Facts []string `json:"facts,omitempty"`
|
|
Concepts []string `json:"concepts,omitempty"`
|
|
FilesRead []string `json:"files_read,omitempty"`
|
|
FilesModified []string `json:"files_modified,omitempty"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.ID == 0 {
|
|
return "", fmt.Errorf("id is required")
|
|
}
|
|
|
|
return s.proxyPutRaw(ctx, fmt.Sprintf("/api/observations/%d", params.ID), params)
|
|
}
|
|
|
|
// handleFindRelatedProxy proxies find related observations.
|
|
func (s *Server) handleFindRelatedProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
ID int64 `json:"id"`
|
|
MinConfidence float64 `json:"min_confidence"`
|
|
Limit int `json:"limit"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.ID == 0 {
|
|
return "", fmt.Errorf("id is required")
|
|
}
|
|
|
|
qp := map[string]string{}
|
|
if params.MinConfidence > 0 {
|
|
qp["min_confidence"] = strconv.FormatFloat(params.MinConfidence, 'f', -1, 64)
|
|
}
|
|
if params.Limit > 0 {
|
|
qp["limit"] = strconv.Itoa(params.Limit)
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, fmt.Sprintf("/api/observations/%d/related", params.ID), qp)
|
|
}
|
|
|
|
// handleFindSimilarProxy proxies find similar observations via context search.
|
|
func (s *Server) handleFindSimilarProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Query string `json:"query"`
|
|
Project string `json:"project"`
|
|
MinSimilarity float64 `json:"min_similarity"`
|
|
Limit int `json:"limit"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.Query == "" {
|
|
return "", fmt.Errorf("query is required")
|
|
}
|
|
if params.Project == "" {
|
|
params.Project = s.project
|
|
}
|
|
if params.Limit == 0 {
|
|
params.Limit = 10
|
|
}
|
|
|
|
// Use context search which uses vector similarity
|
|
return s.proxyGetRaw(ctx, "/api/context/search", map[string]string{
|
|
"project": params.Project,
|
|
"query": params.Query,
|
|
"limit": strconv.Itoa(params.Limit),
|
|
})
|
|
}
|
|
|
|
// handleGetObservationQualityProxy proxies observation quality check.
|
|
// Fetches the observation and computes quality metrics locally (lightweight computation).
|
|
func (s *Server) handleGetObservationQualityProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
ID int64 `json:"id"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.ID == 0 {
|
|
return "", fmt.Errorf("id is required")
|
|
}
|
|
|
|
// Fetch the observation
|
|
obsJSON, err := s.proxyGetRaw(ctx, fmt.Sprintf("/api/observations/%d", params.ID), nil)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
|
|
// Parse observation for quality analysis
|
|
var obs struct {
|
|
Title string `json:"title"`
|
|
Narrative string `json:"narrative"`
|
|
Facts []string `json:"facts"`
|
|
Concepts []string `json:"concepts"`
|
|
FilesRead []string `json:"files_read"`
|
|
FilesModified []string `json:"files_modified"`
|
|
ImportanceScore float64 `json:"importance_score"`
|
|
RetrievalCount int `json:"retrieval_count"`
|
|
IsSuperseded bool `json:"is_superseded"`
|
|
}
|
|
if err := json.Unmarshal([]byte(obsJSON), &obs); err != nil {
|
|
return "", fmt.Errorf("parse observation: %w", err)
|
|
}
|
|
|
|
// Calculate completeness score
|
|
completenessScore := 0.0
|
|
maxScore := 5.0
|
|
var suggestions []string
|
|
|
|
if obs.Title != "" {
|
|
completenessScore += 1.0
|
|
} else {
|
|
suggestions = append(suggestions, "Add a descriptive title")
|
|
}
|
|
|
|
if len(obs.Narrative) > 50 {
|
|
completenessScore += 1.5
|
|
} else if obs.Narrative != "" {
|
|
completenessScore += 0.5
|
|
suggestions = append(suggestions, "Expand the narrative to provide more context (aim for 50+ characters)")
|
|
} else {
|
|
suggestions = append(suggestions, "Add a narrative explaining the observation")
|
|
}
|
|
|
|
if len(obs.Facts) >= 2 {
|
|
completenessScore += 1.0
|
|
} else if len(obs.Facts) == 1 {
|
|
completenessScore += 0.5
|
|
suggestions = append(suggestions, "Add more key facts (aim for 2+)")
|
|
} else {
|
|
suggestions = append(suggestions, "Add key facts to capture important details")
|
|
}
|
|
|
|
if len(obs.Concepts) >= 2 {
|
|
completenessScore += 0.75
|
|
} else if len(obs.Concepts) == 1 {
|
|
completenessScore += 0.25
|
|
suggestions = append(suggestions, "Add more concept tags for better discoverability")
|
|
} else {
|
|
suggestions = append(suggestions, "Add concept tags to categorize this observation")
|
|
}
|
|
|
|
if len(obs.FilesRead) > 0 || len(obs.FilesModified) > 0 {
|
|
completenessScore += 0.75
|
|
} else {
|
|
suggestions = append(suggestions, "Consider adding file references if applicable")
|
|
}
|
|
|
|
qualityTier := "poor"
|
|
switch {
|
|
case completenessScore >= 4.0:
|
|
qualityTier = "excellent"
|
|
case completenessScore >= 3.0:
|
|
qualityTier = "good"
|
|
case completenessScore >= 2.0:
|
|
qualityTier = "fair"
|
|
}
|
|
|
|
response := map[string]any{
|
|
"id": params.ID,
|
|
"completeness_score": completenessScore,
|
|
"max_score": maxScore,
|
|
"completeness_pct": (completenessScore / maxScore) * 100,
|
|
"quality_tier": qualityTier,
|
|
"importance_score": obs.ImportanceScore,
|
|
"retrieval_count": obs.RetrievalCount,
|
|
"is_superseded": obs.IsSuperseded,
|
|
"suggestions": suggestions,
|
|
"field_stats": map[string]any{
|
|
"has_title": obs.Title != "",
|
|
"has_narrative": obs.Narrative != "",
|
|
"narrative_length": len(obs.Narrative),
|
|
"facts_count": len(obs.Facts),
|
|
"concepts_count": len(obs.Concepts),
|
|
"files_read_count": len(obs.FilesRead),
|
|
"files_modified_count": len(obs.FilesModified),
|
|
},
|
|
}
|
|
|
|
output, err := json.Marshal(response)
|
|
if err != nil {
|
|
return "", fmt.Errorf("marshal response: %w", err)
|
|
}
|
|
|
|
return string(output), nil
|
|
}
|
|
|
|
// handleGetRelationshipsProxy proxies observation relationship graph requests.
|
|
func (s *Server) handleGetRelationshipsProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
ID int64 `json:"id"`
|
|
MaxDepth int `json:"max_depth"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid params: %w", err)
|
|
}
|
|
if params.ID <= 0 {
|
|
return "", fmt.Errorf("id is required and must be positive")
|
|
}
|
|
|
|
qp := map[string]string{}
|
|
if params.MaxDepth > 0 {
|
|
qp["max_depth"] = strconv.Itoa(params.MaxDepth)
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, fmt.Sprintf("/api/observations/%d/graph", params.ID), qp)
|
|
}
|
|
|
|
// handleGetScoringBreakdownProxy proxies observation scoring breakdown requests.
|
|
func (s *Server) handleGetScoringBreakdownProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
ID int64 `json:"id"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.ID <= 0 {
|
|
return "", fmt.Errorf("id is required and must be positive")
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, fmt.Sprintf("/api/observations/%d/score", params.ID), nil)
|
|
}
|
|
|
|
// handleTagObservationProxy proxies tag operations on observations.
|
|
// Fetches current tags, computes new tag set, then updates via PUT.
|
|
func (s *Server) handleTagObservationProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Mode string `json:"mode"`
|
|
Tags []string `json:"tags"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.ID == 0 {
|
|
return "", fmt.Errorf("id is required")
|
|
}
|
|
if len(params.Tags) == 0 {
|
|
return "", fmt.Errorf("tags is required")
|
|
}
|
|
if params.Mode == "" {
|
|
params.Mode = "add"
|
|
}
|
|
if params.Mode != "add" && params.Mode != "remove" && params.Mode != "set" {
|
|
return "", fmt.Errorf("mode must be 'add', 'remove', or 'set'")
|
|
}
|
|
|
|
// Fetch current observation to compute new tags
|
|
obsJSON, err := s.proxyGetRaw(ctx, fmt.Sprintf("/api/observations/%d", params.ID), nil)
|
|
if err != nil {
|
|
return "", fmt.Errorf("get observation: %w", err)
|
|
}
|
|
|
|
var obs struct {
|
|
Concepts []string `json:"concepts"`
|
|
}
|
|
if err := json.Unmarshal([]byte(obsJSON), &obs); err != nil {
|
|
return "", fmt.Errorf("parse observation: %w", err)
|
|
}
|
|
|
|
// Compute new tags
|
|
var newTags []string
|
|
switch params.Mode {
|
|
case "set":
|
|
newTags = params.Tags
|
|
case "add":
|
|
tagSet := make(map[string]bool)
|
|
for _, t := range obs.Concepts {
|
|
tagSet[t] = true
|
|
newTags = append(newTags, t)
|
|
}
|
|
for _, t := range params.Tags {
|
|
if !tagSet[t] {
|
|
newTags = append(newTags, t)
|
|
}
|
|
}
|
|
case "remove":
|
|
removeSet := make(map[string]bool)
|
|
for _, t := range params.Tags {
|
|
removeSet[t] = true
|
|
}
|
|
for _, t := range obs.Concepts {
|
|
if !removeSet[t] {
|
|
newTags = append(newTags, t)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Update via PUT
|
|
updatePayload := map[string]any{
|
|
"concepts": newTags,
|
|
}
|
|
result, err := s.proxyPutRaw(ctx, fmt.Sprintf("/api/observations/%d", params.ID), updatePayload)
|
|
if err != nil {
|
|
return "", fmt.Errorf("update observation: %w", err)
|
|
}
|
|
|
|
// Wrap response
|
|
response := map[string]any{
|
|
"id": params.ID,
|
|
"mode": params.Mode,
|
|
"tags_applied": params.Tags,
|
|
"result": json.RawMessage(result),
|
|
}
|
|
output, err := json.Marshal(response)
|
|
if err != nil {
|
|
return "", fmt.Errorf("marshal response: %w", err)
|
|
}
|
|
|
|
return string(output), nil
|
|
}
|
|
|
|
// handleGetObservationsByTagProxy proxies tag-based observation lookup.
|
|
func (s *Server) handleGetObservationsByTagProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Tag string `json:"tag"`
|
|
Project string `json:"project"`
|
|
Limit int `json:"limit"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.Tag == "" {
|
|
return "", fmt.Errorf("tag is required")
|
|
}
|
|
if params.Project == "" {
|
|
params.Project = s.project
|
|
}
|
|
if params.Limit == 0 {
|
|
params.Limit = 50
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, "/api/context/search", map[string]string{
|
|
"project": params.Project,
|
|
"query": params.Tag,
|
|
"concepts": params.Tag,
|
|
"limit": strconv.Itoa(params.Limit),
|
|
})
|
|
}
|
|
|
|
// handleBulkStatusProxy proxies bulk status update operations.
|
|
func (s *Server) handleBulkStatusProxy(ctx context.Context, args json.RawMessage, action string) (string, error) {
|
|
var params struct {
|
|
IDs []int64 `json:"ids"`
|
|
Boost float64 `json:"boost"`
|
|
DeleteVectors bool `json:"delete_vectors"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if len(params.IDs) == 0 {
|
|
return "", fmt.Errorf("ids is required")
|
|
}
|
|
|
|
payload := map[string]any{
|
|
"ids": params.IDs,
|
|
"action": action,
|
|
}
|
|
if action == "boost" {
|
|
payload["boost"] = params.Boost
|
|
}
|
|
if action == "delete" {
|
|
payload["delete_vectors"] = params.DeleteVectors
|
|
}
|
|
|
|
return s.proxyPostRaw(ctx, "/api/observations/bulk-status", payload)
|
|
}
|
|
|
|
// handleMergeProxy proxies merge observations request.
|
|
func (s *Server) handleMergeProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
SourceID int64 `json:"source_id"`
|
|
TargetID int64 `json:"target_id"`
|
|
Boost float64 `json:"boost"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.SourceID == 0 || params.TargetID == 0 {
|
|
return "", fmt.Errorf("source_id and target_id are required")
|
|
}
|
|
|
|
// Merge = mark source as superseded + boost target
|
|
// Step 1: Mark source as superseded
|
|
_, err := s.proxyPostRaw(ctx, "/api/observations/bulk-status", map[string]any{
|
|
"ids": []int64{params.SourceID},
|
|
"action": "supersede",
|
|
})
|
|
if err != nil {
|
|
return "", fmt.Errorf("mark source as superseded: %w", err)
|
|
}
|
|
|
|
// Step 2: Boost target via feedback
|
|
_, err = s.proxyPostRaw(ctx, fmt.Sprintf("/api/observations/%d/feedback", params.TargetID), map[string]any{
|
|
"feedback": "positive",
|
|
})
|
|
if err != nil {
|
|
log.Warn().Err(err).Msg("Failed to boost target observation after merge")
|
|
}
|
|
|
|
response := map[string]any{
|
|
"merged": true,
|
|
"source_id": params.SourceID,
|
|
"target_id": params.TargetID,
|
|
}
|
|
output, err := json.Marshal(response)
|
|
if err != nil {
|
|
return "", fmt.Errorf("marshal response: %w", err)
|
|
}
|
|
|
|
return string(output), nil
|
|
}
|
|
|
|
// handleGetPatternsProxy proxies pattern queries.
|
|
func (s *Server) handleGetPatternsProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Type string `json:"type"`
|
|
Project string `json:"project"`
|
|
Query string `json:"query"`
|
|
Limit int `json:"limit"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
|
|
qp := map[string]string{}
|
|
if params.Type != "" {
|
|
qp["type"] = params.Type
|
|
}
|
|
if params.Project != "" {
|
|
qp["project"] = params.Project
|
|
}
|
|
if params.Limit > 0 {
|
|
qp["limit"] = strconv.Itoa(params.Limit)
|
|
}
|
|
|
|
// Use search endpoint if query provided, otherwise get all
|
|
if params.Query != "" {
|
|
qp["query"] = params.Query
|
|
return s.proxyGetRaw(ctx, "/api/patterns/search", qp)
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, "/api/patterns", qp)
|
|
}
|
|
|
|
// handleAnalyzeImportanceProxy proxies importance analysis.
|
|
func (s *Server) handleAnalyzeImportanceProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Project string `json:"project"`
|
|
Limit int `json:"limit"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.Project == "" {
|
|
params.Project = s.project
|
|
}
|
|
if params.Limit == 0 {
|
|
params.Limit = 10
|
|
}
|
|
|
|
qp := map[string]string{
|
|
"project": params.Project,
|
|
"limit": strconv.Itoa(params.Limit),
|
|
}
|
|
|
|
// Combine results from multiple endpoints
|
|
response := map[string]any{
|
|
"project": params.Project,
|
|
}
|
|
|
|
// Get top-scoring observations
|
|
topScored, err := s.proxyGetRaw(ctx, "/api/observations/top", qp)
|
|
if err == nil {
|
|
response["top_scoring"] = json.RawMessage(topScored)
|
|
}
|
|
|
|
// Get most-retrieved observations
|
|
mostRetrieved, err := s.proxyGetRaw(ctx, "/api/observations/most-retrieved", qp)
|
|
if err == nil {
|
|
response["most_retrieved"] = json.RawMessage(mostRetrieved)
|
|
}
|
|
|
|
// Get scoring stats
|
|
scoringStats, err := s.proxyGetRaw(ctx, "/api/scoring/stats", qp)
|
|
if err == nil {
|
|
response["scoring_stats"] = json.RawMessage(scoringStats)
|
|
}
|
|
|
|
// Get concept weights
|
|
conceptWeights, err := s.proxyGetRaw(ctx, "/api/scoring/concepts", nil)
|
|
if err == nil {
|
|
response["concept_weights"] = json.RawMessage(conceptWeights)
|
|
}
|
|
|
|
output, err := json.Marshal(response)
|
|
if err != nil {
|
|
return "", fmt.Errorf("marshal response: %w", err)
|
|
}
|
|
return string(output), nil
|
|
}
|
|
|
|
// handleExplainSearchProxy proxies search ranking explanation.
|
|
func (s *Server) handleExplainSearchProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Query string `json:"query"`
|
|
Project string `json:"project"`
|
|
TopN int `json:"top_n"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.Query == "" {
|
|
return "", fmt.Errorf("query is required")
|
|
}
|
|
if params.Project == "" {
|
|
params.Project = s.project
|
|
}
|
|
if params.TopN == 0 {
|
|
params.TopN = 5
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, "/api/context/search", map[string]string{
|
|
"project": params.Project,
|
|
"query": params.Query,
|
|
"limit": strconv.Itoa(params.TopN),
|
|
})
|
|
}
|
|
|
|
// handleGetTemporalTrendsProxy proxies temporal trend analysis.
|
|
func (s *Server) handleGetTemporalTrendsProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Project string `json:"project"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.Project == "" {
|
|
params.Project = s.project
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, "/api/stats", map[string]string{
|
|
"project": params.Project,
|
|
})
|
|
}
|
|
|
|
// handleGetDataQualityProxy proxies data quality report.
|
|
func (s *Server) handleGetDataQualityProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Project string `json:"project"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.Project == "" {
|
|
params.Project = s.project
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, "/api/stats", map[string]string{
|
|
"project": params.Project,
|
|
})
|
|
}
|
|
|
|
// handleExportProxy proxies observation export requests.
|
|
func (s *Server) handleExportProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Format string `json:"format"`
|
|
Project string `json:"project"`
|
|
ObsType string `json:"obs_type"`
|
|
Limit int `json:"limit"`
|
|
DateStart int64 `json:"date_start"`
|
|
DateEnd int64 `json:"date_end"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
|
|
qp := map[string]string{}
|
|
if params.Format != "" {
|
|
qp["format"] = params.Format
|
|
}
|
|
if params.Project != "" {
|
|
qp["project"] = params.Project
|
|
} else {
|
|
qp["project"] = s.project
|
|
}
|
|
if params.ObsType != "" {
|
|
qp["obs_type"] = params.ObsType
|
|
}
|
|
if params.Limit > 0 {
|
|
qp["limit"] = strconv.Itoa(params.Limit)
|
|
}
|
|
if params.DateStart > 0 {
|
|
qp["date_start"] = strconv.FormatInt(params.DateStart, 10)
|
|
}
|
|
if params.DateEnd > 0 {
|
|
qp["date_end"] = strconv.FormatInt(params.DateEnd, 10)
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, "/api/observations/export", qp)
|
|
}
|
|
|
|
// handleSuggestConsolidationsProxy proxies consolidation suggestions via duplicates endpoint.
|
|
func (s *Server) handleSuggestConsolidationsProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Project string `json:"project"`
|
|
MinSimilarity float64 `json:"min_similarity"`
|
|
Limit int `json:"limit"`
|
|
}
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.Project == "" {
|
|
params.Project = s.project
|
|
}
|
|
|
|
qp := map[string]string{
|
|
"project": params.Project,
|
|
}
|
|
if params.MinSimilarity > 0 {
|
|
qp["min_similarity"] = strconv.FormatFloat(params.MinSimilarity, 'f', -1, 64)
|
|
}
|
|
if params.Limit > 0 {
|
|
qp["limit"] = strconv.Itoa(params.Limit)
|
|
}
|
|
|
|
return s.proxyGetRaw(ctx, "/api/observations/duplicates", qp)
|
|
}
|
|
|
|
// handleBatchTagProxy proxies batch tag operations.
|
|
// Searches for matching observations and applies tags via PUT endpoint.
|
|
func (s *Server) handleBatchTagProxy(ctx context.Context, args json.RawMessage) (string, error) {
|
|
var params struct {
|
|
Pattern string `json:"pattern"`
|
|
Project string `json:"project"`
|
|
Tags []string `json:"tags"`
|
|
MaxMatches int `json:"max_matches"`
|
|
DryRun bool `json:"dry_run"`
|
|
}
|
|
params.DryRun = true
|
|
params.MaxMatches = 100
|
|
|
|
if err := json.Unmarshal(args, ¶ms); err != nil {
|
|
return "", fmt.Errorf("invalid arguments: %w", err)
|
|
}
|
|
if params.Pattern == "" {
|
|
return "", fmt.Errorf("pattern is required")
|
|
}
|
|
if len(params.Tags) == 0 {
|
|
return "", fmt.Errorf("tags is required")
|
|
}
|
|
if params.Project == "" {
|
|
params.Project = s.project
|
|
}
|
|
|
|
// Search for matching observations
|
|
searchResult, err := s.proxyGetRaw(ctx, "/api/context/search", map[string]string{
|
|
"project": params.Project,
|
|
"query": params.Pattern,
|
|
"limit": strconv.Itoa(params.MaxMatches),
|
|
})
|
|
if err != nil {
|
|
return "", fmt.Errorf("search: %w", err)
|
|
}
|
|
|
|
var searchResp struct {
|
|
Observations []struct {
|
|
ID int64 `json:"id"`
|
|
Title string `json:"title"`
|
|
} `json:"observations"`
|
|
}
|
|
if err := json.Unmarshal([]byte(searchResult), &searchResp); err != nil {
|
|
return "", fmt.Errorf("parse search results: %w", err)
|
|
}
|
|
|
|
matches := make([]map[string]any, 0, len(searchResp.Observations))
|
|
var taggedCount int
|
|
|
|
for _, obs := range searchResp.Observations {
|
|
matches = append(matches, map[string]any{
|
|
"id": obs.ID,
|
|
"title": obs.Title,
|
|
})
|
|
|
|
if !params.DryRun {
|
|
tagArgs, _ := json.Marshal(map[string]any{
|
|
"id": obs.ID,
|
|
"tags": params.Tags,
|
|
"mode": "add",
|
|
})
|
|
_, tagErr := s.handleTagObservationProxy(ctx, tagArgs)
|
|
if tagErr == nil {
|
|
taggedCount++
|
|
}
|
|
}
|
|
}
|
|
|
|
response := map[string]any{
|
|
"pattern": params.Pattern,
|
|
"tags": params.Tags,
|
|
"dry_run": params.DryRun,
|
|
"matches_found": len(matches),
|
|
"matches": matches,
|
|
}
|
|
if !params.DryRun {
|
|
response["tagged_count"] = taggedCount
|
|
}
|
|
|
|
output, err := json.Marshal(response)
|
|
if err != nil {
|
|
return "", fmt.Errorf("marshal response: %w", err)
|
|
}
|
|
|
|
return string(output), nil
|
|
}
|
|
|
|
// sendResponse sends a JSON-RPC response. Returns an error if writing fails.
|
|
func (s *Server) sendResponse(resp *Response) error {
|
|
data, err := json.Marshal(resp)
|
|
if err != nil {
|
|
log.Error().Err(err).Msg("Failed to marshal response")
|
|
// Send a fallback error so the caller isn't left waiting (Fix #3)
|
|
fallback, _ := json.Marshal(map[string]any{
|
|
"jsonrpc": "2.0",
|
|
"id": resp.ID,
|
|
"error": map[string]any{"code": -32603, "message": "internal marshal error"},
|
|
})
|
|
s.writeMu.Lock()
|
|
_, _ = fmt.Fprintln(s.stdout, string(fallback))
|
|
s.writeMu.Unlock()
|
|
return fmt.Errorf("marshal error: %w", err)
|
|
}
|
|
|
|
// Bound the write to prevent pipe deadlock (Fix #2)
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
s.writeMu.Lock()
|
|
_, werr := fmt.Fprintln(s.stdout, string(data))
|
|
s.writeMu.Unlock()
|
|
done <- werr
|
|
}()
|
|
|
|
select {
|
|
case werr := <-done:
|
|
if werr != nil {
|
|
log.Error().Err(werr).Msg("Failed to write response to stdout")
|
|
return werr
|
|
}
|
|
return nil
|
|
case <-time.After(10 * time.Second):
|
|
log.Error().Msg("Stdout write timed out — pipe likely full")
|
|
return fmt.Errorf("write timeout: stdout pipe full")
|
|
}
|
|
}
|
|
|
|
// sendError sends a JSON-RPC error response. Returns an error if writing fails.
|
|
func (s *Server) sendError(id any, code int, message string, data any) error {
|
|
resp := &Response{
|
|
JSONRPC: "2.0",
|
|
ID: id,
|
|
Error: &Error{
|
|
Code: code,
|
|
Message: message,
|
|
Data: data,
|
|
},
|
|
}
|
|
return s.sendResponse(resp)
|
|
}
|