feat: add multi-source drain/MCP aggregation with dedup and source labels

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

Author: killme2008

CLAUDE.md

@@ -51,19 +51,23 @@ AI tool          ←  MCP server (get_build_errors) ←  store.Drain()  ←  per
 ### Key Patterns
 
 - **Multi-adapter fan-out**: Writers discover adapters via `store.DiscoverAdapters()`, write to all. Each tool drains independently.
+- **Multi-source drain**: MCP server and drain command can read from up to 2 sources (local + configured remote). `resolveDrainSources()` in `cmd/devtap/storefactory.go` resolves sources and deduplicates when (backend, session) match. Messages are merged, deduplicated via `mcp.DedupMessages()`, and labeled with `[host session]` prefix in multi-source mode.
+- **Drain budget unit**: `Store.Drain(sessionID, maxLines)` treats `maxLines` as a **message count** (not line count). The multi-source loop tracks remaining budget in messages. Line-level truncation is a separate pass via `mcp.TruncateMessages()`.
+- **Host field**: `LogMessage.Host` is set via cached `os.Hostname()` at capture startup. Used for multi-source origin labeling. GreptimeDB uses `COALESCE(host, '')` for backward compat with pre-existing rows.
 - **File store IPC**: `pending.jsonl` → atomic rename to `pending.jsonl.draining` → read → delete. Leftover lines written back to prevent data loss.
 - **Config merge**: `.mcp.json` / `settings.json` / `opencode.json` reads existing → upserts devtap entry → writes back. Never overwrites other tools' config.
 - **Instruction injection**: Appends `<!-- devtap:start -->` / `<!-- devtap:end -->` block to project instruction files. Idempotent via marker detection.
 - **Session encoding**: `session.EncodeDir("/foo/bar")` → `"-foo-bar"`, shared across adapters.
 - **Capture modes**: `runner.go` (batch, flush every 50 lines) vs `longrun.go` (debounce timer, for dev servers).
 - **Scanner buffers**: 64KB initial / 1MB max (`internal/capture/errors.go`). On scanner error (line >1MB), pipe is drained to discard to prevent child process deadlock.
-- **Line-level truncation**: `mcp.TruncateMessages()` allocates line budget proportionally across messages. Applied in both MCP server and drain command.
+- **Line-level truncation**: `mcp.TruncateMessages()` allocates line budget proportionally across messages. Applied in both MCP server and drain command after all sources are drained.
 
 ### GreptimeDB Specifics
 
 - Composite `PRIMARY KEY (session_id, \`tag\`, stream, adapter)` clause (not inline per-column)
 - `TIMESTAMP(6)` microsecond precision to avoid PK collisions
 - `append_mode=true` allows duplicate PKs
+- `host STRING` column added via `ALTER TABLE ADD COLUMN` (silently ignores error if column already exists). Not a TAG column — TAG columns are part of the primary key and cannot be added via ALTER TABLE.
 - SQL injection protection via `validateFilterSQL` in drain (best-effort blocklist, not a security boundary)
 - Integration tests gated behind `//go:build integration`
 
@@ -72,10 +76,19 @@ AI tool          ←  MCP server (get_build_errors) ←  store.Drain()  ←  per
 
 devtap captures build/dev output from a separate terminal and delivers it here via MCP.
 
-**You MUST call get_build_errors in these situations:**
+**You MUST check devtap in these situations:**
 - At the start of each new task or conversation turn
 - Before writing or editing any code
 - After being told that a build, test, or dev server has new output
 
-Do not assume there are no errors — always check. Build output may arrive at any time from a parallel terminal.
+**Use this call order:**
+1. Call "get_build_status" once at the start of the turn.
+2. Call "get_build_errors" when status reports pending messages, when the user says new output arrived, or before the first code edit in the turn.
+
+**Call limit per turn:** call "get_build_errors" at most once per turn unless new output is reported.
+
+**Multi-source mode:** when devtap drains from multiple sources, tags are prefixed with "host/label |" (for example, `[devtap: myhost/local | make]`). "host" is the machine name, "label" identifies the source. Show these prefixes as-is. If output includes source warnings (for example, source unreachable), show those warnings verbatim and continue with output from reachable sources.
+
+**Output format:** when "get_build_errors" returns content, present it verbatim in a fenced code block, then add one line:
+"Next action: <what you will do>".
 <!-- devtap:end -->

README.md

@@ -60,6 +60,19 @@ devtap -- npm run dev
 
 **Terminal B** — use your AI coding tool as usual. It will automatically call `get_build_errors` via MCP to fetch captured build errors.
 
+If you want to verify without MCP, run:
+
+```bash
+devtap drain
+```
+
+Typical output:
+
+```text
+[devtap: cargo check] Build failed (exit code 101):
+...
+```
+
 **Tip:** Since devtap captures stdout from any command, you can send arbitrary messages to your coding agent:
 
 ```bash
@@ -106,6 +119,13 @@ Your laptop                       CI / remote build server
 3. Each AI tool independently drains its own copy via `get_build_errors`
 4. AI sees the errors and fixes them
 
+When `mcp-serve`/`drain` is started with explicit `--session` or `--store`, devtap can merge output from two sources:
+
+- `local`: auto-detected project session from your default backend
+- `configured`: the explicit `--session`/`--store` target
+
+If both resolve to the same backend+session, devtap uses a single source. Otherwise it drains both, deduplicates identical messages, and prefixes tags with source info (for example `myhost/local |`).
+
 ## Supported Tools
 
 | Tool | Adapter | Integration | Config File | Instruction File |
@@ -218,6 +238,30 @@ devtap --store greptimedb --session myproject -- make
 
 Multiple build machines can write to the same session simultaneously — each entry is tagged with its source, and the AI tool drains them all.
 
+**Local + remote merged drain** — keep your local loop and a remote shared session visible at the same time:
+
+```bash
+# Laptop: local output (default source)
+devtap -- cargo check
+
+# Remote machine: shared session output (configured source)
+devtap --store greptimedb --session myproject -- make
+
+# Laptop: install MCP with explicit remote session
+devtap install --adapter claude-code --store greptimedb --session myproject
+
+# Optional manual check (without MCP client)
+devtap drain --store greptimedb --session myproject
+```
+
+`devtap drain` shows a merged header (`Draining from N sources`), emits source-unreachable warnings when partial failures occur, and still returns available output from reachable sources.
+
+Typical merged header:
+
+```text
+[devtap] Draining from 2 sources (2 reachable)
+```
+
 **Session auto-detection** — when `--session auto` (default), devtap resolves the project directory like this:
 1. Git root (nearest parent with `.git`)
 2. Project marker files (nearest parent with one of: `go.mod`, `package.json`, `pyproject.toml`, `Cargo.toml`, `pom.xml`, `build.gradle`, `build.gradle.kts`, `composer.json`, `Gemfile`, `setup.py`)
@@ -265,6 +309,15 @@ Subcommands:
 
 Lines exceeding `--max-lines` are smart-truncated: head and tail preserved with omission notice. Consecutive duplicate lines are merged.
 
+`devtap mcp-serve` and `devtap drain` can aggregate multiple sources (local + configured) as described above. `devtap drain --filter-sql` is a single-source mode and requires `--store greptimedb`.
+
+## Troubleshooting
+
+- No output but you expect logs: run `devtap status` first, then `devtap drain --max-lines 200`.
+- Using an unexpected session: run with `--session pick` once to confirm the target session.
+- Multi-source drain shows warnings: reachable sources are still returned; warnings indicate one source is unavailable.
+- MCP tool not being called: re-run `devtap install --adapter <adapter>` in the project root and restart the AI tool session.
+
 ## Security & Privacy
 
 - **All data stays local.** Build output is stored on your machine at `~/.devtap/` (file backend) or in a self-hosted GreptimeDB instance. Nothing is sent to external servers.

cmd/devtap/drain.go

@@ -3,6 +3,7 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
 
 	"github.com/spf13/cobra"
 
@@ -33,14 +34,107 @@ For MCP-capable tools, use "devtap mcp-serve" instead.`,
 }
 
 func runDrain(cmd *cobra.Command, args []string) error {
-	adapterName, _ := cmd.Flags().GetString("adapter")
-	sessionFlag, _ := cmd.Flags().GetString("session")
 	event, _ := cmd.Flags().GetString("event")
 	autoLoop, _ := cmd.Flags().GetBool("auto-loop")
 	maxRetries, _ := cmd.Flags().GetInt("max-retries")
 	maxLines, _ := cmd.Flags().GetInt("max-lines")
 	filterSQL, _ := cmd.Flags().GetString("filter-sql")
 
+	// If --filter-sql is set, fall back to single-source drain against the configured store.
+	if filterSQL != "" {
+		return runDrainSingleSource(cmd, filterSQL, maxLines, event, autoLoop, maxRetries)
+	}
+
+	sources, cleanup, err := resolveDrainSources(cmd)
+	if err != nil {
+		return fmt.Errorf("resolve drain sources: %w", err)
+	}
+	defer cleanup()
+
+	multiSource := len(sources) > 1
+	var allMessages []store.LogMessage
+
+	// Track remaining message budget. Drain treats its limit as a message
+	// count, so the budget uses the same unit. Line-level truncation is
+	// handled afterward by TruncateMessages.
+	remaining := maxLines
+
+	for _, src := range sources {
+		if remaining <= 0 {
+			break
+		}
+
+		messages, err := src.Store.Drain(src.SessionID, remaining)
+		if err != nil {
+			if multiSource {
+				fmt.Fprintf(cmd.ErrOrStderr(), "devtap: source %q unreachable: %v\n", src.Label, err)
+				continue
+			}
+			return fmt.Errorf("drain: %w", err)
+		}
+
+		if multiSource {
+			for i := range messages {
+				host := messages[i].Host
+				if host == "" {
+					host = "unknown"
+				}
+				messages[i].Tag = fmt.Sprintf("%s/%s | %s", host, src.Label, messages[i].Tag)
+			}
+		}
+
+		allMessages = append(allMessages, messages...)
+		remaining -= len(messages)
+	}
+
+	if multiSource && len(allMessages) > 0 {
+		allMessages = mcp.DedupMessages(allMessages)
+	}
+
+	allMessages = mcp.TruncateMessages(allMessages, maxLines)
+
+	// Handle auto-loop Stop hook (Claude Code specific)
+	if event == "Stop" && autoLoop {
+		storeDir, err := defaultStoreDir()
+		if err != nil {
+			return fmt.Errorf("resolve store dir: %w", err)
+		}
+		// Use the first source's session for retry tracking
+		sessionID := sources[0].SessionID
+		return handleAutoLoopStop(storeDir, sessionID, allMessages, maxRetries)
+	}
+
+	// Reset retry counter on any non-Stop drain (e.g., user submitted a new prompt).
+	if event != "" && event != "Stop" {
+		if storeDir, err := defaultStoreDir(); err == nil {
+			tracker := store.NewRetryTracker(storeDir)
+			for _, src := range sources {
+				_ = tracker.Reset(src.SessionID)
+			}
+		}
+	}
+
+	if len(allMessages) == 0 {
+		return nil
+	}
+
+	// Plain text output
+	if multiSource {
+		var sourceLabels []string
+		for _, src := range sources {
+			sourceLabels = append(sourceLabels, src.Label)
+		}
+		fmt.Printf("[devtap] Draining from %d sources: %s\n\n", len(sourceLabels), strings.Join(sourceLabels, ", "))
+	}
+	fmt.Println(mcp.FormatMessages(allMessages))
+	return nil
+}
+
+// runDrainSingleSource handles drain with --filter-sql which only works with a single GreptimeDB store.
+func runDrainSingleSource(cmd *cobra.Command, filterSQL string, maxLines int, event string, autoLoop bool, maxRetries int) error {
+	adapterName, _ := cmd.Flags().GetString("adapter")
+	sessionFlag, _ := cmd.Flags().GetString("session")
+
 	sessionID, err := resolveSession(adapterName, sessionFlag)
 	if err != nil {
 		return fmt.Errorf("resolve session: %w", err)
@@ -52,24 +146,18 @@ func runDrain(cmd *cobra.Command, args []string) error {
 	}
 	defer func() { _ = s.Close() }()
 
-	// Drain messages (with SQL filter if using GreptimeDB)
-	var messages []store.LogMessage
-	if filterSQL != "" {
-		if gs, ok := s.(*greptimestore.Store); ok {
-			messages, err = gs.DrainSQL(sessionID, filterSQL, maxLines)
-		} else {
-			return fmt.Errorf("--filter-sql requires --store greptimedb")
-		}
-	} else {
-		messages, err = s.Drain(sessionID, maxLines)
+	gs, ok := s.(*greptimestore.Store)
+	if !ok {
+		return fmt.Errorf("--filter-sql requires --store greptimedb")
 	}
+
+	messages, err := gs.DrainSQL(sessionID, filterSQL, maxLines)
 	if err != nil {
 		return fmt.Errorf("drain: %w", err)
 	}
 
 	messages = mcp.TruncateMessages(messages, maxLines)
 
-	// Handle auto-loop Stop hook (Claude Code specific)
 	if event == "Stop" && autoLoop {
 		storeDir, err := defaultStoreDir()
 		if err != nil {
@@ -78,7 +166,6 @@ func runDrain(cmd *cobra.Command, args []string) error {
 		return handleAutoLoopStop(storeDir, sessionID, messages, maxRetries)
 	}
 
-	// Reset retry counter on any non-Stop drain (e.g., user submitted a new prompt).
 	if event != "" && event != "Stop" {
 		if storeDir, err := defaultStoreDir(); err == nil {
 			tracker := store.NewRetryTracker(storeDir)
@@ -90,7 +177,6 @@ func runDrain(cmd *cobra.Command, args []string) error {
 		return nil
 	}
 
-	// Plain text output
 	fmt.Println(mcp.FormatMessages(messages))
 	return nil
 }

cmd/devtap/install.go

@@ -59,6 +59,15 @@ func runInstall(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("install: %w", err)
 	}
 
+	// Print multi-source hint when explicit --session or --store is used.
+	if (sessionFlag != "" && sessionFlag != "auto") || storeFlag != "" {
+		targetSession := sessionFlag
+		if targetSession == "" {
+			targetSession = "auto"
+		}
+		fmt.Printf("MCP server will drain from both local builds and remote session %q.\n\n", targetSession)
+	}
+
 	switch adapterName {
 	case "claude-code":
 		fmt.Println("Installed devtap MCP server for Claude Code.")

cmd/devtap/mcpserve.go

@@ -23,38 +23,30 @@ Exposed tools:
   get_build_status   Get pending message counts per session`,
 		SilenceUsage: true,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			s, err := openStore(cmd)
-			if err != nil {
-				return fmt.Errorf("open store: %w", err)
+			maxLines, _ := cmd.Flags().GetInt("max-lines")
+			if maxLines <= 0 {
+				maxLines = 10000
 			}
-			defer func() { _ = s.Close() }()
 
 			adapterName, _ := cmd.Flags().GetString("adapter")
 			if adapterName == "" {
 				adapterName = "claude-code"
 			}
-			sessionID, _ := cmd.Flags().GetString("session")
-			maxLines, _ := cmd.Flags().GetInt("max-lines")
-			if maxLines <= 0 {
-				maxLines = 10000
-			}
 
-			// If session is "auto", use a default based on cwd
-			if sessionID == "" || sessionID == "auto" {
-				resolved, err := resolveSession(adapterName, "auto")
-				if err != nil || resolved == "" {
-					sessionID = "default"
-				} else {
-					sessionID = resolved
-				}
+			sources, cleanup, err := resolveDrainSources(cmd)
+			if err != nil {
+				return fmt.Errorf("resolve drain sources: %w", err)
 			}
+			defer cleanup()
 
-			// Pre-register adapter dir so writers can discover and fan-out to us.
+			// Pre-register adapter dir for each source so writers can discover us.
 			if storeDir, err := defaultStoreDir(); err == nil {
-				_ = store.EnsureAdapterDir(storeDir, sessionID, adapterName)
+				for _, src := range sources {
+					_ = store.EnsureAdapterDir(storeDir, src.SessionID, adapterName)
+				}
 			}
 
-			srv := mcp.NewServer(s, sessionID, maxLines)
+			srv := mcp.NewMultiSourceServer(sources, maxLines)
 			return srv.Run()
 		},
 	}

cmd/devtap/run.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 
 	"github.com/spf13/cobra"
 
@@ -37,10 +38,9 @@ func runRun(cmd *cobra.Command, args []string) error {
 	debounce, _ := cmd.Flags().GetDuration("debounce")
 
 	if tag == "" {
-		if name := capture.CommandName(cmdArgs); name != "" {
-			tag = filepath.Base(name)
-		} else {
-			tag = filepath.Base(cmdArgs[0])
+		tag = shellJoin(cmdArgs)
+		if len(tag) > 80 {
+			tag = tag[:77] + "..."
 		}
 	}
 
@@ -175,3 +175,17 @@ func findMarkerRoot(start string) (string, bool) {
 	}
 	return "", false
 }
+
+// shellJoin joins command arguments into a display string,
+// quoting arguments that contain spaces or shell metacharacters.
+func shellJoin(args []string) string {
+	parts := make([]string, len(args))
+	for i, arg := range args {
+		if strings.ContainsAny(arg, " \t'\"\\$`!#&|;(){}") {
+			parts[i] = fmt.Sprintf("%q", arg)
+		} else {
+			parts[i] = arg
+		}
+	}
+	return strings.Join(parts, " ")
+}