feat: add source field to knowledge insights, unify connections tool (#113)

* feat: add source field to knowledge insights for agent-discovered knowledge

Add a source field to capture_insight that distinguishes user-provided
insights from agent-discovered ones. Three source values: user (default),
agent_discovery (agent figured it out via data exploration), and
enrichment_gap (metadata gap flagged for admin attention).

- Migration 000010: add source column with DEFAULT 'user' for backward compat
- Types: source constants, ValidateSource, NormalizeSource
- Store: INSERT/SELECT/filter support for source field
- Toolkit: input schema, validation, tool description updated
- Prompt: expanded agent guidance for self-capture and when to ask users
- Docs: overview, admin-api, governance, llms.txt, llms-full.txt updated
- Swagger: regenerated for new Insight.Source field

* test: fix patch coverage for PR #113

Add targeted tests covering previously-uncovered code paths:

- datahub/toolkit: RegisterTools with non-nil inner toolkit
- trino/toolkit: RegisterTools with non-nil inner toolkit
- platform/connections_tool: invoke tool through MCP transport
- platform/lifecycle: rollback with nil stop callback
- knowledge/toolkit: MarkApplied error path in apply handler
- oauth/postgres: cleanup goroutine error handling
- oauth/server: cleanup routine error handling

Patch coverage: 92.4% (134/145 changed lines)

Author: cjimti

cmd/mcp-data-platform/main.go

@@ -80,7 +80,6 @@ func setupSignalHandler() context.Context {
 type serverResult struct {
 	mcpServer *mcp.Server
 	platform  *platform.Platform
-	toolkit   interface{ Close() error }
 }
 
 func createServer(opts serverOptions) (*serverResult, error) {
@@ -95,14 +94,32 @@ func createServer(opts serverOptions) (*serverResult, error) {
 		return result, nil
 	}
 
-	result.mcpServer, result.toolkit, err = mcpserver.NewWithDefaults()
+	result.mcpServer, err = mcpserver.NewWithDefaults()
 	if err != nil {
 		return nil, fmt.Errorf("creating server with defaults: %w", err)
 	}
 	return result, nil
 }
 
+// initLogging configures slog from the LOG_LEVEL environment variable.
+// Supported values: debug, info, warn, error. Defaults to info.
+func initLogging() {
+	level := slog.LevelInfo
+	switch os.Getenv("LOG_LEVEL") {
+	case "debug", "DEBUG":
+		level = slog.LevelDebug
+	case "warn", "WARN":
+		level = slog.LevelWarn
+	case "error", "ERROR":
+		level = slog.LevelError
+	}
+	slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{
+		Level: level,
+	})))
+}
+
 func run() error {
+	initLogging()
 	opts := parseFlags()
 
 	if opts.showVersion {
@@ -129,11 +146,6 @@ func closeServer(result *serverResult) {
 			slog.Error("shutdown: platform close error", "error", err)
 		}
 	}
-	if result.toolkit != nil {
-		if err := result.toolkit.Close(); err != nil {
-			slog.Error("shutdown: toolkit close error", "error", err)
-		}
-	}
 	slog.Info("shutdown: complete")
 }
 

cmd/mcp-data-platform/main_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"log/slog"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -23,6 +24,40 @@ const (
 	testPreDelay3s     = 3 * time.Second
 )
 
+func TestInitLogging(t *testing.T) {
+	tests := []struct {
+		env   string
+		level slog.Level
+	}{
+		{"debug", slog.LevelDebug},
+		{"DEBUG", slog.LevelDebug},
+		{"info", slog.LevelInfo},
+		{"warn", slog.LevelWarn},
+		{"WARN", slog.LevelWarn},
+		{"error", slog.LevelError},
+		{"ERROR", slog.LevelError},
+		{"", slog.LevelInfo},        // default
+		{"unknown", slog.LevelInfo}, // unrecognized falls through
+	}
+
+	for _, tt := range tests {
+		t.Run("LOG_LEVEL="+tt.env, func(t *testing.T) {
+			t.Setenv("LOG_LEVEL", tt.env)
+			initLogging()
+
+			handler := slog.Default().Handler()
+			// Verify the handler is enabled at the expected level
+			if !handler.Enabled(context.Background(), tt.level) {
+				t.Errorf("expected handler enabled at %v", tt.level)
+			}
+			// For non-debug levels, debug should be disabled
+			if tt.level > slog.LevelDebug && handler.Enabled(context.Background(), slog.LevelDebug) {
+				t.Errorf("expected debug disabled when LOG_LEVEL=%q", tt.env)
+			}
+		})
+	}
+}
+
 func TestRegisterOAuthRoutes(t *testing.T) {
 	mux := http.NewServeMux()
 	handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {

docs/knowledge/admin-api.md

@@ -34,6 +34,7 @@ Returns a paginated list of insights with optional filtering.
 | `entity_urn` | string | Filter by related entity URN |
 | `captured_by` | string | Filter by the user who captured the insight |
 | `confidence` | string | Filter by confidence level: `high`, `medium`, `low` |
+| `source` | string | Filter by source: `user`, `agent_discovery`, `enrichment_gap` |
 | `since` | RFC 3339 | Filter insights created after this timestamp |
 | `until` | RFC 3339 | Filter insights created before this timestamp |
 | `page` | integer | Page number, 1-based (default: 1) |
@@ -57,6 +58,7 @@ curl -s "https://mcp.example.com/api/v1/admin/knowledge/insights?status=pending&
       "session_id": "sess_abc123",
       "captured_by": "analyst@example.com",
       "persona": "analyst",
+      "source": "user",
       "category": "correction",
       "insight_text": "The amount column represents gross margin before returns, not revenue.",
       "confidence": "high",
@@ -348,6 +350,7 @@ Knowledge capture uses two PostgreSQL tables, created by migrations 000006, 0000
 | `session_id` | TEXT | MCP session that produced the insight |
 | `captured_by` | TEXT | User who shared the knowledge |
 | `persona` | TEXT | Active persona at capture time |
+| `source` | TEXT | Where the knowledge came from: `user`, `agent_discovery`, `enrichment_gap` |
 | `category` | TEXT | Insight category |
 | `insight_text` | TEXT | The domain knowledge content |
 | `confidence` | TEXT | Confidence level (high, medium, low) |

docs/knowledge/governance.md

@@ -265,7 +265,7 @@ sequenceDiagram
     Analyst->>AI: That's gross margin before returns,<br/>not revenue like the name suggests
 
     Note over AI,Platform: Capture
-    AI->>Platform: capture_insight(<br/>category: correction,<br/>entity_urns: [urn:li:dataset:...orders...],<br/>insight_text: "amount column is gross margin<br/>before returns, not revenue",<br/>confidence: high,<br/>suggested_actions: [{<br/>  action_type: update_description,<br/>  target: amount,<br/>  detail: "Gross margin before returns"<br/>}])
+    AI->>Platform: capture_insight(<br/>category: correction,<br/>source: user,<br/>entity_urns: [urn:li:dataset:...orders...],<br/>insight_text: "amount column is gross margin<br/>before returns, not revenue",<br/>confidence: high,<br/>suggested_actions: [{<br/>  action_type: update_description,<br/>  target: amount,<br/>  detail: "Gross margin before returns"<br/>}])
     Platform->>DB: INSERT (status: pending)
     Platform-->>AI: Insight captured: a1b2c3
 

docs/knowledge/overview.md

@@ -174,11 +174,23 @@ personas:
           - "apply_knowledge"
 ```
 
+## Insight Sources
+
+Insights track where the knowledge came from via the `source` field:
+
+| Source | Description | Example |
+|--------|-------------|---------|
+| `user` | Knowledge shared by the user during conversation (default) | User says "The amount column is gross margin, not revenue" |
+| `agent_discovery` | Knowledge the AI agent figured out independently | Agent samples data and discovers a column contains ISO country codes |
+| `enrichment_gap` | Metadata gap flagged for admin attention | Table has no description and the agent cannot determine its purpose from the data |
+
+The source field is optional when calling `capture_insight`. When omitted, it defaults to `user`.
+
 ## AI Agent Guidance
 
 The toolkit registers an MCP prompt called `knowledge_capture_guidance` that tells AI assistants when to capture insights. The prompt covers:
 
-**When to capture:**
+**When to capture (user-provided):**
 
 - User corrects a column description, table purpose, or data interpretation
 - User explains what data means in business terms not captured in metadata
@@ -187,12 +199,29 @@ The toolkit registers an MCP prompt called `knowledge_capture_guidance` that tel
 - User explains connections between datasets not captured in lineage
 - User suggests improvements to existing documentation or metadata
 
+**When to capture (agent-discovered):**
+
+- Agent discovers what a column means by sampling actual data (set `source: "agent_discovery"`)
+- Agent finds join relationships not documented in lineage metadata
+- Agent identifies data quality patterns (nulls, outliers, encoding issues)
+- Agent resolves ambiguous column names by examining values
+- Agent encounters metadata that is missing or clearly wrong and cannot resolve it from the data (set `source: "enrichment_gap"`)
+
+**When to ask the user instead:**
+
+- Enrichment is insufficient and the agent cannot resolve it from the data alone
+- Multiple interpretations are equally plausible
+- The insight would have high impact (e.g., PII classification, deprecation status)
+
 **When not to capture:**
 
 - Transient questions or debugging ("why is my query slow?")
 - Personal preferences ("I prefer using CTEs")
 - Information already present in the catalog metadata
 - Vague or unverifiable claims without specific context
+- Trivially obvious gaps without adding what the data actually means
+- Speculative interpretations without evidence from querying
+- The same gap repeatedly within a session
 
 The prompt is available via `prompts/list` and `prompts/get` in the MCP protocol.
 

docs/llms-full.txt

@@ -987,6 +987,7 @@ Record domain knowledge shared during a session.
 | `category` | string | Yes | - | correction, business_context, data_quality, usage_guidance, relationship, enhancement |
 | `insight_text` | string | Yes | - | Knowledge to record (10-4000 chars) |
 | `confidence` | string | No | medium | high, medium, low |
+| `source` | string | No | user | user, agent_discovery, enrichment_gap |
 | `entity_urns` | array | No | [] | Related DataHub entity URNs (max 10) |
 | `related_columns` | array | No | [] | Related columns (max 20) |
 | `suggested_actions` | array | No | [] | Proposed catalog changes (max 5) |
@@ -1154,6 +1155,7 @@ Records domain knowledge shared during a session.
 | `category` | string | Yes | One of: `correction`, `business_context`, `data_quality`, `usage_guidance`, `relationship`, `enhancement` |
 | `insight_text` | string | Yes | The knowledge to record (10-4000 characters) |
 | `confidence` | string | No | `high`, `medium` (default), or `low` |
+| `source` | string | No | `user` (default), `agent_discovery`, or `enrichment_gap` |
 | `entity_urns` | array | No | DataHub URNs this insight relates to (max 10) |
 | `related_columns` | array | No | Columns related to this insight (max 20) |
 | `suggested_actions` | array | No | Proposed catalog changes (max 5) |
@@ -1310,6 +1312,7 @@ HTTP endpoints for managing insights and changesets. All endpoints require admin
 | `entity_urn` | string | Filter by entity URN |
 | `captured_by` | string | Filter by user who captured |
 | `confidence` | string | Filter by confidence level |
+| `source` | string | Filter by source (user, agent_discovery, enrichment_gap) |
 | `since` | RFC3339 | Filter by creation time (after) |
 | `until` | RFC3339 | Filter by creation time (before) |
 | `page` | integer | Page number (1-based) |
@@ -1348,6 +1351,7 @@ Knowledge capture uses two PostgreSQL tables (migrations 000006, 000007, 000008)
 | `session_id` | TEXT | MCP session that produced it |
 | `captured_by` | TEXT | User who shared the knowledge |
 | `persona` | TEXT | Active persona at capture time |
+| `source` | TEXT | Where the knowledge came from: `user`, `agent_discovery`, `enrichment_gap` |
 | `category` | TEXT | Insight category |
 | `insight_text` | TEXT | The domain knowledge |
 | `confidence` | TEXT | Confidence level |

docs/llms.txt

@@ -53,7 +53,7 @@
 
 ## Knowledge Capture
 
-- [Overview](knowledge/overview.md): Tribal knowledge capture for data catalogs. capture_insight records domain knowledge during AI sessions; apply_knowledge provides admin review, synthesis, and DataHub write-back with changeset tracking and rollback. Insight categories, lifecycle states, governance workflow, configuration, persona integration, AI agent guidance prompt
+- [Overview](knowledge/overview.md): Tribal knowledge capture for data catalogs. capture_insight records domain knowledge during AI sessions; apply_knowledge provides admin review, synthesis, and DataHub write-back with changeset tracking and rollback. Insight categories, lifecycle states, insight sources (user, agent_discovery, enrichment_gap), governance workflow, configuration, persona integration, AI agent guidance prompt for both user-provided and agent-discovered insights
 - [Governance Workflow](knowledge/governance.md): Active metadata management through human-in-the-loop curation. Bulk review, per-entity review, approve/reject actions, synthesize change proposals, apply changes to DataHub, changeset tracking, rollback. Complete end-to-end workflow example
 - [Admin API](knowledge/admin-api.md): REST endpoints for managing insights and changesets. List/filter/get/update insights, approve/reject status transitions, insight statistics. List/get changesets, rollback. Authentication, error responses, database schema reference
 

internal/apidocs/docs.go

@@ -2614,6 +2614,9 @@ const docTemplate = `{
                 "session_id": {
                     "type": "string"
                 },
+                "source": {
+                    "type": "string"
+                },
                 "status": {
                     "type": "string"
                 },

internal/apidocs/swagger.json

@@ -2608,6 +2608,9 @@
                 "session_id": {
                     "type": "string"
                 },
+                "source": {
+                    "type": "string"
+                },
                 "status": {
                     "type": "string"
                 },

internal/apidocs/swagger.yaml

@@ -559,6 +559,8 @@ definitions:
         type: string
       session_id:
         type: string
+      source:
+        type: string
       status:
         type: string
       suggested_actions: