fix(mcp): fail-closed process ownership, interrupted-write reconnect, stricter redaction

commit: e6d4fe0882d69b73d1b37f87616b79e4c429e48d
author: brice · 7d ago · 2026-10-15 13:51 UTC
parent: be3ffc0
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

11 files · +838 -131

M packages/agent/mcp/fakeserver_test.go +21 -2


@@ -29,7 +29,7 @@ import (
29	29		const fakeServerEnv = "ZUT_MCP_FAKE_SERVER"
30	30
31	31		// fakeToolCount is the number of tools the fake server lists in total.
32		-	const fakeToolCount = 17
	32	+	const fakeToolCount = 18
33	33
34	34		var fakeImageBytes = []byte("\x89PNG\r\n\x1a\nfake-image-bytes")
35	35

@@ -58,7 +58,10 @@ type fakeBackend struct {
58	58		extraTool atomic.Bool
59	59		// manyResources makes resources/list return this many long entries.
60	60		manyResources int
61		-	events string
	61	+	// hugeDescription, when > 0, gives the first resource a description of
	62	+	// that many bytes.
	63	+	hugeDescription int
	64	+	events string
62	65		}
63	66
64	67		func fakeBackendFromEnv() *fakeBackend {

@@ -71,6 +74,9 @@ func fakeBackendFromEnv() *fakeBackend {
71	74		if n, err := strconv.Atoi(os.Getenv("FAKE_MANY_RESOURCES")); err == nil {
72	75		b.manyResources = n
73	76		}
	77	+	if n, err := strconv.Atoi(os.Getenv("FAKE_HUGE_DESC")); err == nil {
	78	+	b.hugeDescription = n
	79	+	}
74	80		return b
75	81		}
76	82

@@ -118,6 +124,7 @@ func (b *fakeBackend) toolPage(cursor string) map[string]any {
118	124		fakeToolDef("spawn", "Start a descendant process and return its pid", false),
119	125		fakeToolDef("stall", "Stop reading stdin", false),
120	126		fakeToolDef("stderr_flood", "Write a huge stderr line", false),
	127	+	fakeToolDef("stderr_secret", "Write a stderr line with FAKE_SECRET straddling byte 4096", false),
121	128		fakeToolDef("cwd", "Return the server working directory", false),
122	129		}
123	130		if b.extraTool.Load() {

@@ -211,6 +218,11 @@ func (b fakeBackend) handle(method string, params json.RawMessage) (any, RPCEr
211	218		return nil, notFound
212	219		}
213	220		fakeEvent(b.events, "list-resources")
	221	+	if b.hugeDescription > 0 {
	222	+	return map[string]any{"resources": []any{map[string]any{
	223	+	"uri": "mem://huge", "name": "huge", "description": strings.Repeat("h", b.hugeDescription),
	224	+	}, map[string]any{"uri": "mem://after", "name": "after"}}}, nil
	225	+	}
214	226		if b.manyResources > 0 {
215	227		list := make([]any, 0, b.manyResources)
216	228		for i := 0; i < b.manyResources; i++ {

@@ -367,6 +379,13 @@ func runFakeStdioServer() {
367	379		}
368	380		respond(msg.ID, fakeCallResult(fakeText(strconv.Itoa(child.Process.Pid))), nil)
369	381		return
	382	+	case "stderr_secret":
	383	+	secret := os.Getenv("FAKE_SECRET")
	384	+	pad := 4096 - len(secret)/2
	385	+	fmt.Fprintln(os.Stderr, strings.Repeat("x", pad)+secret+strings.Repeat("y", 200))
	386	+	fmt.Fprintln(os.Stderr, "after-secret")
	387	+	respond(msg.ID, fakeCallResult(fakeText("written")), nil)
	388	+	return
370	389		case "stderr_flood":
371	390		os.Stderr.WriteString(strings.Repeat("x", 3<<20) + "\n")
372	391		fmt.Fprintln(os.Stderr, "after-flood")

 const fakeServerEnv = "ZUT_MCP_FAKE_SERVER"
 // fakeToolCount is the number of tools the fake server lists in total.
-const fakeToolCount = 17
+const fakeToolCount = 18
 var fakeImageBytes = []byte("\x89PNG\r\n\x1a\nfake-image-bytes")
 	extraTool   atomic.Bool
 	// manyResources makes resources/list return this many long entries.
 	manyResources int
-	events        string
+	// hugeDescription, when > 0, gives the first resource a description of
+	// that many bytes.
+	hugeDescription int
+	events          string
 }
 func fakeBackendFromEnv() *fakeBackend {
 	if n, err := strconv.Atoi(os.Getenv("FAKE_MANY_RESOURCES")); err == nil {
 		b.manyResources = n
 	}
+	if n, err := strconv.Atoi(os.Getenv("FAKE_HUGE_DESC")); err == nil {
+		b.hugeDescription = n
+	}
 	return b
 }
 		fakeToolDef("spawn", "Start a descendant process and return its pid", false),
 		fakeToolDef("stall", "Stop reading stdin", false),
 		fakeToolDef("stderr_flood", "Write a huge stderr line", false),
+		fakeToolDef("stderr_secret", "Write a stderr line with FAKE_SECRET straddling byte 4096", false),
 		fakeToolDef("cwd", "Return the server working directory", false),
 	}
 	if b.extraTool.Load() {
 			return nil, notFound
 		}
 		fakeEvent(b.events, "list-resources")
+		if b.hugeDescription > 0 {
+			return map[string]any{"resources": []any{map[string]any{
+				"uri": "mem://huge", "name": "huge", "description": strings.Repeat("h", b.hugeDescription),
+			}, map[string]any{"uri": "mem://after", "name": "after"}}}, nil
+		}
 		if b.manyResources > 0 {
 			list := make([]any, 0, b.manyResources)
 			for i := 0; i < b.manyResources; i++ {
 				}
 				respond(msg.ID, fakeCallResult(fakeText(strconv.Itoa(child.Process.Pid))), nil)
 				return
+			case "stderr_secret":
+				secret := os.Getenv("FAKE_SECRET")
+				pad := 4096 - len(secret)/2
+				fmt.Fprintln(os.Stderr, strings.Repeat("x", pad)+secret+strings.Repeat("y", 200))
+				fmt.Fprintln(os.Stderr, "after-secret")
+				respond(msg.ID, fakeCallResult(fakeText("written")), nil)
+				return
 			case "stderr_flood":
 				os.Stderr.WriteString(strings.Repeat("x", 3<<20) + "\n")
 				fmt.Fprintln(os.Stderr, "after-flood")

M packages/agent/mcp/hardening_test.go +16 -6


@@ -558,20 +558,30 @@ func TestStdioCancelNotificationRules(t *testing.T) {
558	558		if got := st.cancelsSent.Load(); got != 0 {
559	559		t.Errorf("notifications/cancelled sent for initialize (%d)", got)
560	560		}
561		-	// Any other method is announced exactly once, without blocking the caller.
562		-	ctx2, cancel2 := context.WithCancel(context.Background())
563		-	cancel2()
	561	+	}
	562	+
	563	+	func TestStdioCancelNotificationForSentRequest(t *testing.T) {
	564	+	c, events := connectFake(t, nil)
	565	+	st := c.t.(*stdioTransport)
	566	+	ctx, cancel := context.WithCancel(context.Background())
	567	+	done := make(chan error, 1)
	568	+	go func() {
	569	+	_, err := c.CallTool(ctx, "slow", []byte(`{}`))
	570	+	done <- err
	571	+	}()
	572	+	waitEvent(t, events, "slow-started")
564	573		start := time.Now()
565		-	_, err = st.request(ctx2, 2, "tools/call", map[string]any{"name": "slow", "arguments": map[string]any{}})
566		-	if !errors.Is(err, context.Canceled) {
	574	+	cancel()
	575	+	if err := <-done; !errors.Is(err, context.Canceled) {
567	576		t.Fatalf("err = %v", err)
568	577		}
569	578		if time.Since(start) > time.Second {
570	579		t.Error("caller delayed by the cancel notification")
571	580		}
572	581		if got := st.cancelsSent.Load(); got != 1 {
573		-	t.Errorf("cancel count for tools/call = %d, want 1", got)
	582	+	t.Errorf("cancel count = %d, want 1", got)
574	583		}
	584	+	waitEvent(t, events, "cancelled:")
575	585		}
576	586
577	587		func TestHTTPInitializeNotCancelledOnTimeout(t *testing.T) {


@@ -558,20 +558,30 @@ func TestStdioCancelNotificationRules(t *testing.T) {
558		if got := st.cancelsSent.Load(); got != 0 {	558		if got := st.cancelsSent.Load(); got != 0 {
559		t.Errorf("notifications/cancelled sent for initialize (%d)", got)	559		t.Errorf("notifications/cancelled sent for initialize (%d)", got)
560		}	560		}
561	-	// Any other method is announced exactly once, without blocking the caller.	561	+	}
562	-	ctx2, cancel2 := context.WithCancel(context.Background())	562	+
563	-	cancel2()	563	+	func TestStdioCancelNotificationForSentRequest(t *testing.T) {
			564	+	c, events := connectFake(t, nil)
			565	+	st := c.t.(*stdioTransport)
			566	+	ctx, cancel := context.WithCancel(context.Background())
			567	+	done := make(chan error, 1)
			568	+	go func() {
			569	+	_, err := c.CallTool(ctx, "slow", []byte(`{}`))
			570	+	done <- err
			571	+	}()
			572	+	waitEvent(t, events, "slow-started")
564		start := time.Now()	573		start := time.Now()
565	-	_, err = st.request(ctx2, 2, "tools/call", map[string]any{"name": "slow", "arguments": map[string]any{}})	574	+	cancel()
566	-	if !errors.Is(err, context.Canceled) {	575	+	if err := <-done; !errors.Is(err, context.Canceled) {
567		t.Fatalf("err = %v", err)	576		t.Fatalf("err = %v", err)
568		}	577		}
569		if time.Since(start) > time.Second {	578		if time.Since(start) > time.Second {
570		t.Error("caller delayed by the cancel notification")	579		t.Error("caller delayed by the cancel notification")
571		}	580		}
572		if got := st.cancelsSent.Load(); got != 1 {	581		if got := st.cancelsSent.Load(); got != 1 {
573	-	t.Errorf("cancel count for tools/call = %d, want 1", got)	582	+	t.Errorf("cancel count = %d, want 1", got)
574		}	583		}
			584	+	waitEvent(t, events, "cancelled:")
575		}	585		}
576			586
577		func TestHTTPInitializeNotCancelledOnTimeout(t *testing.T) {	587		func TestHTTPInitializeNotCancelledOnTimeout(t *testing.T) {

M packages/agent/mcp/http.go +2 -1


@@ -95,7 +95,8 @@ func (t *httpTransport) newRequest(ctx context.Context, method string, body []by
95	95		}
96	96		req, err := http.NewRequestWithContext(ctx, method, t.url, reader)
97	97		if err != nil {
98		-	return nil, err
	98	+	// url parse errors quote the whole URL, query token included.
	99	+	return nil, t.red.scrubError(err)
99	100		}
100	101		for k, v := range t.headers {
101	102		req.Header.Set(k, v)

 	}
 	req, err := http.NewRequestWithContext(ctx, method, t.url, reader)
 	if err != nil {
-		return nil, err
+		// url parse errors quote the whole URL, query token included.
+		return nil, t.red.scrubError(err)
 	}
 	for k, v := range t.headers {
 		req.Header.Set(k, v)

M packages/agent/mcp/manager_test.go +2 -2


@@ -160,14 +160,14 @@ func TestManagerStdioToolsAndDescriptions(t *testing.T) {
160	160		}
161	161
162	162		text := FormatStatus(m.Status())
163		-	if !strings.Contains(text, "fake (stdio): connected, 17 tools, resources") {
	163	+	if !strings.Contains(text, "fake (stdio): connected, 18 tools, resources") {
164	164		t.Errorf("FormatStatus = %q", text)
165	165		}
166	166		if FormatStatus(nil) != "no MCP servers configured" {
167	167		t.Error("empty FormatStatus")
168	168		}
169	169		add := m.SystemPromptAddendum()
170		-	for _, want := range []string{"[mcp_servers]", "fake: 17 tools", "mcp__fake__<tool>", "Be nice"} {
	170	+	for _, want := range []string{"[mcp_servers]", "fake: 18 tools", "mcp__fake__<tool>", "Be nice"} {
171	171		if !strings.Contains(add, want) {
172	172		t.Errorf("addendum missing %q:\n%s", want, add)
173	173		}


@@ -160,14 +160,14 @@ func TestManagerStdioToolsAndDescriptions(t *testing.T) {
160		}	160		}
161			161
162		text := FormatStatus(m.Status())	162		text := FormatStatus(m.Status())
163	-	if !strings.Contains(text, "fake (stdio): connected, 17 tools, resources") {	163	+	if !strings.Contains(text, "fake (stdio): connected, 18 tools, resources") {
164		t.Errorf("FormatStatus = %q", text)	164		t.Errorf("FormatStatus = %q", text)
165		}	165		}
166		if FormatStatus(nil) != "no MCP servers configured" {	166		if FormatStatus(nil) != "no MCP servers configured" {
167		t.Error("empty FormatStatus")	167		t.Error("empty FormatStatus")
168		}	168		}
169		add := m.SystemPromptAddendum()	169		add := m.SystemPromptAddendum()
170	-	for _, want := range []string{"[mcp_servers]", "fake: 17 tools", "mcp__fake__<tool>", "Be nice"} {	170	+	for _, want := range []string{"[mcp_servers]", "fake: 18 tools", "mcp__fake__<tool>", "Be nice"} {
171		if !strings.Contains(add, want) {	171		if !strings.Contains(add, want) {
172		t.Errorf("addendum missing %q:\n%s", want, add)	172		t.Errorf("addendum missing %q:\n%s", want, add)
173		}	173		}

M packages/agent/mcp/proc_unix.go +95 -17


@@ -3,49 +3,127 @@
3	3		package mcp
4	4
5	5		import (
	6	+	"errors"
	7	+	"os"
6	8		"os/exec"
	9	+	"sync"
7	10		"syscall"
	11	+	"time"
8	12		)
9	13
10	14		// isolateProcess puts the server in its own process group so it does not
11		-	// receive terminal signals aimed at zut and so close can stop its children.
	15	+	// receive terminal signals aimed at zut and so cleanup can stop its children.
12	16		func isolateProcess(cmd *exec.Cmd) {
13	17		cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
14	18		}
15	19
16		-	// processGuard tracks everything a stdio server spawned: the process group on
17		-	// Unix, a Job Object on Windows.
	20	+	// processGuard owns the process group of one stdio server.
	21	+	//
	22	+	// A numeric process-group id is only safe to signal while the group still has
	23	+	// members: once the last member is gone the kernel may hand the number to an
	24	+	// unrelated process. The guard therefore (a) cleans the group exactly once,
	25	+	// (b) forgets the id as soon as the group is observed empty or cleaned, and
	26	+	// (c) is cleaned as soon as the direct child is reaped, so no stale id stays
	27	+	// signalable for the life of the transport.
18	28		type processGuard struct {
19		-	cmd *exec.Cmd
	29	+	mu sync.Mutex
	30	+	pgid int // 0 once cleaned up or observed gone
	31	+	proc *os.Process
	32	+	kill func(pid int, sig syscall.Signal) error // syscall.Kill; replaced in tests
	33	+
	34	+	once sync.Once
20	35		}
21	36
22	37		// newProcessGuard is called right after cmd.Start.
23		-	func newProcessGuard(cmd exec.Cmd) processGuard { return &processGuard{cmd: cmd} }
	38	+	func newProcessGuard(cmd exec.Cmd) (processGuard, error) {
	39	+	g := &processGuard{kill: func(pid int, sig syscall.Signal) error { return syscall.Kill(pid, sig) }}
	40	+	if cmd != nil && cmd.Process != nil {
	41	+	g.pgid = cmd.Process.Pid
	42	+	g.proc = cmd.Process
	43	+	}
	44	+	return g, nil
	45	+	}
24	46
25		-	// terminate signals the whole process group, including descendants that
26		-	// outlive the direct child.
	47	+	// terminate signals the whole group, including descendants that outlive the
	48	+	// direct child. It does nothing once the group is cleaned up.
27	49		func (g *processGuard) terminate(force bool) {
28		-	if g == nil \|\| g.cmd == nil \|\| g.cmd.Process == nil {
	50	+	if g == nil {
	51	+	return
	52	+	}
	53	+	g.mu.Lock()
	54	+	defer g.mu.Unlock()
	55	+	if g.pgid == 0 {
29	56		return
30	57		}
31	58		sig := syscall.SIGTERM
32	59		if force {
33	60		sig = syscall.SIGKILL
34	61		}
35		-	err := syscall.Kill(-g.cmd.Process.Pid, sig)
36		-	if err != nil && err != syscall.ESRCH {
37		-	_ = g.cmd.Process.Signal(sig)
	62	+	err := g.kill(-g.pgid, sig)
	63	+	switch {
	64	+	case err == nil:
	65	+	case errors.Is(err, syscall.ESRCH):
	66	+	g.pgid = 0 // group gone: never signal this number again
	67	+	default:
	68	+	if g.proc != nil {
	69	+	_ = g.proc.Signal(sig)
	70	+	}
38	71		}
39	72		}
40	73
41		-	// alive reports whether any member of the process group still exists.
	74	+	// alive reports whether the group still has members.
42	75		func (g *processGuard) alive() bool {
43		-	if g == nil \|\| g.cmd == nil \|\| g.cmd.Process == nil {
	76	+	if g == nil {
	77	+	return false
	78	+	}
	79	+	g.mu.Lock()
	80	+	defer g.mu.Unlock()
	81	+	if g.pgid == 0 {
44	82		return false
45	83		}
46		-	err := syscall.Kill(-g.cmd.Process.Pid, 0)
47		-	return err == nil \|\| err == syscall.EPERM
	84	+	err := g.kill(-g.pgid, 0)
	85	+	if errors.Is(err, syscall.ESRCH) {
	86	+	g.pgid = 0
	87	+	return false
	88	+	}
	89	+	return true // nil, or EPERM: a member exists that we cannot signal
48	90		}
49	91
50		-	// release frees resources held for the group. Nothing is held on Unix.
51		-	func (g *processGuard) release() {}
	92	+	// cleanup stops every remaining group member (SIGTERM, then SIGKILL for
	93	+	// survivors) and releases the id. It runs its body once; later and concurrent
	94	+	// calls wait for that run and then return.
	95	+	func (g *processGuard) cleanup() {
	96	+	if g == nil {
	97	+	return
	98	+	}
	99	+	g.once.Do(func() {
	100	+	defer func() {
	101	+	g.mu.Lock()
	102	+	g.pgid = 0
	103	+	g.proc = nil
	104	+	g.mu.Unlock()
	105	+	}()
	106	+	if !g.alive() {
	107	+	return
	108	+	}
	109	+	g.terminate(false)
	110	+	if g.waitGone(300 * time.Millisecond) {
	111	+	return
	112	+	}
	113	+	g.terminate(true)
	114	+	g.waitGone(200 * time.Millisecond)
	115	+	})
	116	+	}
	117	+
	118	+	func (g *processGuard) waitGone(d time.Duration) bool {
	119	+	deadline := time.Now().Add(d)
	120	+	for {
	121	+	if !g.alive() {
	122	+	return true
	123	+	}
	124	+	if time.Now().After(deadline) {
	125	+	return false
	126	+	}
	127	+	time.Sleep(10 * time.Millisecond)
	128	+	}
	129	+	}


@@ -3,49 +3,127 @@
3		package mcp	3		package mcp
4			4
5		import (	5		import (
			6	+	"errors"
			7	+	"os"
6		"os/exec"	8		"os/exec"
			9	+	"sync"
7		"syscall"	10		"syscall"
			11	+	"time"
8		)	12		)
9			13
10		// isolateProcess puts the server in its own process group so it does not	14		// isolateProcess puts the server in its own process group so it does not
11	-	// receive terminal signals aimed at zut and so close can stop its children.	15	+	// receive terminal signals aimed at zut and so cleanup can stop its children.
12		func isolateProcess(cmd *exec.Cmd) {	16		func isolateProcess(cmd *exec.Cmd) {
13		cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}	17		cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
14		}	18		}
15			19
16	-	// processGuard tracks everything a stdio server spawned: the process group on	20	+	// processGuard owns the process group of one stdio server.
17	-	// Unix, a Job Object on Windows.	21	+	//
			22	+	// A numeric process-group id is only safe to signal while the group still has
			23	+	// members: once the last member is gone the kernel may hand the number to an
			24	+	// unrelated process. The guard therefore (a) cleans the group exactly once,
			25	+	// (b) forgets the id as soon as the group is observed empty or cleaned, and
			26	+	// (c) is cleaned as soon as the direct child is reaped, so no stale id stays
			27	+	// signalable for the life of the transport.
18		type processGuard struct {	28		type processGuard struct {
19	-	cmd *exec.Cmd	29	+	mu sync.Mutex
			30	+	pgid int // 0 once cleaned up or observed gone
			31	+	proc *os.Process
			32	+	kill func(pid int, sig syscall.Signal) error // syscall.Kill; replaced in tests
			33	+
			34	+	once sync.Once
20		}	35		}
21			36
22		// newProcessGuard is called right after cmd.Start.	37		// newProcessGuard is called right after cmd.Start.
23	-	func newProcessGuard(cmd exec.Cmd) processGuard { return &processGuard{cmd: cmd} }	38	+	func newProcessGuard(cmd exec.Cmd) (processGuard, error) {
			39	+	g := &processGuard{kill: func(pid int, sig syscall.Signal) error { return syscall.Kill(pid, sig) }}
			40	+	if cmd != nil && cmd.Process != nil {
			41	+	g.pgid = cmd.Process.Pid
			42	+	g.proc = cmd.Process
			43	+	}
			44	+	return g, nil
			45	+	}
24			46
25	-	// terminate signals the whole process group, including descendants that	47	+	// terminate signals the whole group, including descendants that outlive the
26	-	// outlive the direct child.	48	+	// direct child. It does nothing once the group is cleaned up.
27		func (g *processGuard) terminate(force bool) {	49		func (g *processGuard) terminate(force bool) {
28	-	if g == nil \|\| g.cmd == nil \|\| g.cmd.Process == nil {	50	+	if g == nil {
			51	+	return
			52	+	}
			53	+	g.mu.Lock()
			54	+	defer g.mu.Unlock()
			55	+	if g.pgid == 0 {
29		return	56		return
30		}	57		}
31		sig := syscall.SIGTERM	58		sig := syscall.SIGTERM
32		if force {	59		if force {
33		sig = syscall.SIGKILL	60		sig = syscall.SIGKILL
34		}	61		}
35	-	err := syscall.Kill(-g.cmd.Process.Pid, sig)	62	+	err := g.kill(-g.pgid, sig)
36	-	if err != nil && err != syscall.ESRCH {	63	+	switch {
37	-	_ = g.cmd.Process.Signal(sig)	64	+	case err == nil:
			65	+	case errors.Is(err, syscall.ESRCH):
			66	+	g.pgid = 0 // group gone: never signal this number again
			67	+	default:
			68	+	if g.proc != nil {
			69	+	_ = g.proc.Signal(sig)
			70	+	}
38		}	71		}
39		}	72		}
40			73
41	-	// alive reports whether any member of the process group still exists.	74	+	// alive reports whether the group still has members.
42		func (g *processGuard) alive() bool {	75		func (g *processGuard) alive() bool {
43	-	if g == nil \|\| g.cmd == nil \|\| g.cmd.Process == nil {	76	+	if g == nil {
			77	+	return false
			78	+	}
			79	+	g.mu.Lock()
			80	+	defer g.mu.Unlock()
			81	+	if g.pgid == 0 {
44		return false	82		return false
45		}	83		}
46	-	err := syscall.Kill(-g.cmd.Process.Pid, 0)	84	+	err := g.kill(-g.pgid, 0)
47	-	return err == nil \|\| err == syscall.EPERM	85	+	if errors.Is(err, syscall.ESRCH) {
			86	+	g.pgid = 0
			87	+	return false
			88	+	}
			89	+	return true // nil, or EPERM: a member exists that we cannot signal
48		}	90		}
49			91
50	-	// release frees resources held for the group. Nothing is held on Unix.	92	+	// cleanup stops every remaining group member (SIGTERM, then SIGKILL for
51	-	func (g *processGuard) release() {}	93	+	// survivors) and releases the id. It runs its body once; later and concurrent
			94	+	// calls wait for that run and then return.
			95	+	func (g *processGuard) cleanup() {
			96	+	if g == nil {
			97	+	return
			98	+	}
			99	+	g.once.Do(func() {
			100	+	defer func() {
			101	+	g.mu.Lock()
			102	+	g.pgid = 0
			103	+	g.proc = nil
			104	+	g.mu.Unlock()
			105	+	}()
			106	+	if !g.alive() {
			107	+	return
			108	+	}
			109	+	g.terminate(false)
			110	+	if g.waitGone(300 * time.Millisecond) {
			111	+	return
			112	+	}
			113	+	g.terminate(true)
			114	+	g.waitGone(200 * time.Millisecond)
			115	+	})
			116	+	}
			117	+
			118	+	func (g *processGuard) waitGone(d time.Duration) bool {
			119	+	deadline := time.Now().Add(d)
			120	+	for {
			121	+	if !g.alive() {
			122	+	return true
			123	+	}
			124	+	if time.Now().After(deadline) {
			125	+	return false
			126	+	}
			127	+	time.Sleep(10 * time.Millisecond)
			128	+	}
			129	+	}

A packages/agent/mcp/proc_unix_guard_test.go +135 -0

@@ -0,0 +1,135 @@
	1	+	//go:build !windows
	2	+
	3	+	package mcp
	4	+
	5	+	import (
	6	+	"strconv"
	7	+	"sync"
	8	+	"syscall"
	9	+	"testing"
	10	+	)
	11	+
	12	+	// fakeKill records signals and answers from a script.
	13	+	type fakeKill struct {
	14	+	mu sync.Mutex
	15	+	calls []int // pids passed, in order (negative = group)
	16	+	err error
	17	+	}
	18	+
	19	+	func (f *fakeKill) kill(pid int, _ syscall.Signal) error {
	20	+	f.mu.Lock()
	21	+	defer f.mu.Unlock()
	22	+	f.calls = append(f.calls, pid)
	23	+	return f.err
	24	+	}
	25	+
	26	+	func (f *fakeKill) count() int {
	27	+	f.mu.Lock()
	28	+	defer f.mu.Unlock()
	29	+	return len(f.calls)
	30	+	}
	31	+
	32	+	func newGuardForTest(pgid int, k fakeKill) processGuard {
	33	+	return &processGuard{pgid: pgid, kill: k.kill}
	34	+	}
	35	+
	36	+	func TestGuardForgetsGroupOnceGone(t *testing.T) {
	37	+	k := &fakeKill{err: syscall.ESRCH}
	38	+	g := newGuardForTest(4242, k)
	39	+	if g.alive() {
	40	+	t.Fatal("ESRCH means the group is gone")
	41	+	}
	42	+	before := k.count()
	43	+	// A gone group is never probed or signalled again: its number may belong
	44	+	// to somebody else by now.
	45	+	g.terminate(false)
	46	+	g.terminate(true)
	47	+	if g.alive() {
	48	+	t.Error("alive after gone")
	49	+	}
	50	+	g.cleanup()
	51	+	if k.count() != before {
	52	+	t.Errorf("signals sent to a group already observed gone: %v", k.calls)
	53	+	}
	54	+	}
	55	+
	56	+	func TestGuardCleanupIsOneShot(t *testing.T) {
	57	+	k := &fakeKill{} // every probe/signal "succeeds": the group never empties
	58	+	g := newGuardForTest(4243, k)
	59	+	g.cleanup() // escalates SIGTERM then SIGKILL, then gives up
	60	+	after := k.count()
	61	+	if after == 0 {
	62	+	t.Fatal("cleanup sent nothing to a live group")
	63	+	}
	64	+	for i := 0; i < 3; i++ {
	65	+	g.cleanup()
	66	+	g.terminate(true)
	67	+	g.alive()
	68	+	}
	69	+	if k.count() != after {
	70	+	t.Errorf("signals after cleanup: %d more", k.count()-after)
	71	+	}
	72	+	if g.pgid != 0 {
	73	+	t.Error("process group id retained after cleanup")
	74	+	}
	75	+	}
	76	+
	77	+	func TestGuardNoSignalWhenAlreadyEmpty(t *testing.T) {
	78	+	k := &fakeKill{err: syscall.ESRCH}
	79	+	g := newGuardForTest(4244, k)
	80	+	g.cleanup()
	81	+	for _, pid := range k.calls {
	82	+	if pid != -4244 {
	83	+	t.Errorf("unexpected target %d", pid)
	84	+	}
	85	+	}
	86	+	// Only the liveness probe ran, and nothing after it.
	87	+	if k.count() != 1 {
	88	+	t.Errorf("calls = %v, want a single probe", k.calls)
	89	+	}
	90	+	}
	91	+
	92	+	func TestStdioRepeatedCloseDoesNotSignalAgain(t *testing.T) {
	93	+	cfg, _ := fakeStdioConfig(t, nil)
	94	+	st, err := startStdio("x", cfg, "")
	95	+	if err != nil {
	96	+	t.Fatal(err)
	97	+	}
	98	+	if err := st.close(); err != nil {
	99	+	t.Fatal(err)
	100	+	}
	101	+	k := &fakeKill{}
	102	+	st.guard.mu.Lock()
	103	+	st.guard.kill = k.kill
	104	+	st.guard.mu.Unlock()
	105	+	for i := 0; i < 3; i++ {
	106	+	if err := st.close(); err != nil {
	107	+	t.Fatal(err)
	108	+	}
	109	+	}
	110	+	if k.count() != 0 {
	111	+	t.Errorf("late Close signalled a saved process-group id: %v", k.calls)
	112	+	}
	113	+	}
	114	+
	115	+	func TestStdioNaturalExitCleansDescendants(t *testing.T) {
	116	+	c, _ := connectFake(t, nil)
	117	+	pid, err := strconv.Atoi(callText(t, c, "spawn", `{"mode":"sleeper-stubborn"}`))
	118	+	if err != nil {
	119	+	t.Fatal(err)
	120	+	}
	121	+	t.Cleanup(func() { killPID(pid) })
	122	+	if !processAlive(pid) {
	123	+	t.Fatal("descendant not running")
	124	+	}
	125	+	// The server dies on its own (no Close): the descendant must not be left
	126	+	// running, and the transport must have released the group by itself.
	127	+	_, _ = c.CallTool(t.Context(), "exit", []byte(`{}`))
	128	+	eventually(t, "descendant to be cleaned after the leader exited", func() bool { return !processAlive(pid) })
	129	+	st := c.t.(*stdioTransport)
	130	+	eventually(t, "group id to be released", func() bool {
	131	+	st.guard.mu.Lock()
	132	+	defer st.guard.mu.Unlock()
	133	+	return st.guard.pgid == 0
	134	+	})
	135	+	}

@@ -0,0 +1,135 @@
	1	+	//go:build !windows
	2	+
	3	+	package mcp
	4	+
	5	+	import (
	6	+	"strconv"
	7	+	"sync"
	8	+	"syscall"
	9	+	"testing"
	10	+	)
	11	+
	12	+	// fakeKill records signals and answers from a script.
	13	+	type fakeKill struct {
	14	+	mu sync.Mutex
	15	+	calls []int // pids passed, in order (negative = group)
	16	+	err error
	17	+	}
	18	+
	19	+	func (f *fakeKill) kill(pid int, _ syscall.Signal) error {
	20	+	f.mu.Lock()
	21	+	defer f.mu.Unlock()
	22	+	f.calls = append(f.calls, pid)
	23	+	return f.err
	24	+	}
	25	+
	26	+	func (f *fakeKill) count() int {
	27	+	f.mu.Lock()
	28	+	defer f.mu.Unlock()
	29	+	return len(f.calls)
	30	+	}
	31	+
	32	+	func newGuardForTest(pgid int, k fakeKill) processGuard {
	33	+	return &processGuard{pgid: pgid, kill: k.kill}
	34	+	}
	35	+
	36	+	func TestGuardForgetsGroupOnceGone(t *testing.T) {
	37	+	k := &fakeKill{err: syscall.ESRCH}
	38	+	g := newGuardForTest(4242, k)
	39	+	if g.alive() {
	40	+	t.Fatal("ESRCH means the group is gone")
	41	+	}
	42	+	before := k.count()
	43	+	// A gone group is never probed or signalled again: its number may belong
	44	+	// to somebody else by now.
	45	+	g.terminate(false)
	46	+	g.terminate(true)
	47	+	if g.alive() {
	48	+	t.Error("alive after gone")
	49	+	}
	50	+	g.cleanup()
	51	+	if k.count() != before {
	52	+	t.Errorf("signals sent to a group already observed gone: %v", k.calls)
	53	+	}
	54	+	}
	55	+
	56	+	func TestGuardCleanupIsOneShot(t *testing.T) {
	57	+	k := &fakeKill{} // every probe/signal "succeeds": the group never empties
	58	+	g := newGuardForTest(4243, k)
	59	+	g.cleanup() // escalates SIGTERM then SIGKILL, then gives up
	60	+	after := k.count()
	61	+	if after == 0 {
	62	+	t.Fatal("cleanup sent nothing to a live group")
	63	+	}
	64	+	for i := 0; i < 3; i++ {
	65	+	g.cleanup()
	66	+	g.terminate(true)
	67	+	g.alive()
	68	+	}
	69	+	if k.count() != after {
	70	+	t.Errorf("signals after cleanup: %d more", k.count()-after)
	71	+	}
	72	+	if g.pgid != 0 {
	73	+	t.Error("process group id retained after cleanup")
	74	+	}
	75	+	}
	76	+
	77	+	func TestGuardNoSignalWhenAlreadyEmpty(t *testing.T) {
	78	+	k := &fakeKill{err: syscall.ESRCH}
	79	+	g := newGuardForTest(4244, k)
	80	+	g.cleanup()
	81	+	for _, pid := range k.calls {
	82	+	if pid != -4244 {
	83	+	t.Errorf("unexpected target %d", pid)
	84	+	}
	85	+	}
	86	+	// Only the liveness probe ran, and nothing after it.
	87	+	if k.count() != 1 {
	88	+	t.Errorf("calls = %v, want a single probe", k.calls)
	89	+	}
	90	+	}
	91	+
	92	+	func TestStdioRepeatedCloseDoesNotSignalAgain(t *testing.T) {
	93	+	cfg, _ := fakeStdioConfig(t, nil)
	94	+	st, err := startStdio("x", cfg, "")
	95	+	if err != nil {
	96	+	t.Fatal(err)
	97	+	}
	98	+	if err := st.close(); err != nil {
	99	+	t.Fatal(err)
	100	+	}
	101	+	k := &fakeKill{}
	102	+	st.guard.mu.Lock()
	103	+	st.guard.kill = k.kill
	104	+	st.guard.mu.Unlock()
	105	+	for i := 0; i < 3; i++ {
	106	+	if err := st.close(); err != nil {
	107	+	t.Fatal(err)
	108	+	}
	109	+	}
	110	+	if k.count() != 0 {
	111	+	t.Errorf("late Close signalled a saved process-group id: %v", k.calls)
	112	+	}
	113	+	}
	114	+
	115	+	func TestStdioNaturalExitCleansDescendants(t *testing.T) {
	116	+	c, _ := connectFake(t, nil)
	117	+	pid, err := strconv.Atoi(callText(t, c, "spawn", `{"mode":"sleeper-stubborn"}`))
	118	+	if err != nil {
	119	+	t.Fatal(err)
	120	+	}
	121	+	t.Cleanup(func() { killPID(pid) })
	122	+	if !processAlive(pid) {
	123	+	t.Fatal("descendant not running")
	124	+	}
	125	+	// The server dies on its own (no Close): the descendant must not be left
	126	+	// running, and the transport must have released the group by itself.
	127	+	_, _ = c.CallTool(t.Context(), "exit", []byte(`{}`))
	128	+	eventually(t, "descendant to be cleaned after the leader exited", func() bool { return !processAlive(pid) })
	129	+	st := c.t.(*stdioTransport)
	130	+	eventually(t, "group id to be released", func() bool {
	131	+	st.guard.mu.Lock()
	132	+	defer st.guard.mu.Unlock()
	133	+	return st.guard.pgid == 0
	134	+	})
	135	+	}

M packages/agent/mcp/proc_windows.go +95 -45


@@ -3,90 +3,140 @@
3	3		package mcp
4	4
5	5		import (
	6	+	"errors"
	7	+	"fmt"
6	8		"os/exec"
7	9		"sync"
	10	+	"syscall"
8	11		"unsafe"
9	12
10	13		"golang.org/x/sys/windows"
11	14		)
12	15
13		-	func isolateProcess(_ *exec.Cmd) {}
	16	+	// isolateProcess starts the server suspended. The guard assigns it to a Job
	17	+	// Object before its first instruction runs and then resumes it, so no
	18	+	// descendant can be created outside the job.
	19	+	func isolateProcess(cmd *exec.Cmd) {
	20	+	if cmd.SysProcAttr == nil {
	21	+	cmd.SysProcAttr = &syscall.SysProcAttr{}
	22	+	}
	23	+	cmd.SysProcAttr.CreationFlags \|= windows.CREATE_SUSPENDED
	24	+	}
14	25
15		-	// processGuard assigns the server to a Job Object with
16		-	// JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, so descendants die with the server (and
17		-	// with zut, should zut itself be killed).
18		-	//
19		-	// Assignment happens right after Start; a server that spawns children in the
20		-	// first instants of its life can escape the job. If the job cannot be created
21		-	// or assigned, the guard falls back to killing the direct child.
	26	+	// processGuard owns a Job Object with JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE that
	27	+	// contains the server and all its descendants. The mutex is held across every
	28	+	// use of the handle so a concurrent release cannot close it mid-call.
22	29		type processGuard struct {
23		-	cmd *exec.Cmd
24		-
25		-	mu sync.Mutex
26		-	job windows.Handle
	30	+	mu sync.Mutex
	31	+	job windows.Handle // 0 once released
	32	+	once sync.Once
27	33		}
28	34
29		-	func newProcessGuard(cmd exec.Cmd) processGuard {
30		-	g := &processGuard{cmd: cmd}
	35	+	// newProcessGuard is called right after cmd.Start with the process still
	36	+	// suspended. It fails closed: if the job cannot be set up, the child is killed
	37	+	// and an error is returned.
	38	+	func newProcessGuard(cmd exec.Cmd) (processGuard, error) {
31	39		if cmd == nil \|\| cmd.Process == nil {
32		-	return g
	40	+	return nil, errors.New("process not started")
	41	+	}
	42	+	fail := func(what string, err error) (*processGuard, error) {
	43	+	_ = cmd.Process.Kill()
	44	+	return nil, fmt.Errorf("%s: %w", what, err)
33	45		}
34	46		job, err := windows.CreateJobObject(nil, nil)
35	47		if err != nil {
36		-	return g
	48	+	return fail("create job object", err)
37	49		}
38	50		var info windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION
39	51		info.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
40	52		if _, err := windows.SetInformationJobObject(job, windows.JobObjectExtendedLimitInformation,
41	53		uintptr(unsafe.Pointer(&info)), uint32(unsafe.Sizeof(info))); err != nil {
42	54		_ = windows.CloseHandle(job)
43		-	return g
	55	+	return fail("configure job object", err)
44	56		}
45	57		proc, err := windows.OpenProcess(windows.PROCESS_SET_QUOTA\|windows.PROCESS_TERMINATE, false, uint32(cmd.Process.Pid))
46	58		if err != nil {
47	59		_ = windows.CloseHandle(job)
48		-	return g
	60	+	return fail("open server process", err)
49	61		}
50		-	defer windows.CloseHandle(proc)
51		-	if err := windows.AssignProcessToJobObject(job, proc); err != nil {
	62	+	err = windows.AssignProcessToJobObject(job, proc)
	63	+	_ = windows.CloseHandle(proc)
	64	+	if err != nil {
52	65		_ = windows.CloseHandle(job)
53		-	return g
	66	+	return fail("assign server to job object", err)
54	67		}
55		-	g.job = job
56		-	return g
	68	+	if err := resumeProcess(uint32(cmd.Process.Pid)); err != nil {
	69	+	_ = windows.CloseHandle(job) // kills the job's only member
	70	+	return fail("resume server", err)
	71	+	}
	72	+	return &processGuard{job: job}, nil
57	73		}
58	74
59		-	// terminate kills the job (every process in it) or, without a job, the direct
60		-	// child. Windows has no graceful group signal, so force is ignored.
	75	+	// resumeProcess resumes every suspended thread of pid (a CREATE_SUSPENDED
	76	+	// process has exactly its main thread).
	77	+	func resumeProcess(pid uint32) error {
	78	+	snap, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPTHREAD, 0)
	79	+	if err != nil {
	80	+	return err
	81	+	}
	82	+	defer windows.CloseHandle(snap)
	83	+	entry := windows.ThreadEntry32{Size: uint32(unsafe.Sizeof(windows.ThreadEntry32{}))}
	84	+	if err := windows.Thread32First(snap, &entry); err != nil {
	85	+	return err
	86	+	}
	87	+	resumed := 0
	88	+	for {
	89	+	if entry.OwnerProcessID == pid {
	90	+	th, err := windows.OpenThread(windows.THREAD_SUSPEND_RESUME, false, entry.ThreadID)
	91	+	if err != nil {
	92	+	return err
	93	+	}
	94	+	_, err = windows.ResumeThread(th)
	95	+	_ = windows.CloseHandle(th)
	96	+	if err != nil {
	97	+	return err
	98	+	}
	99	+	resumed++
	100	+	}
	101	+	entry.Size = uint32(unsafe.Sizeof(entry))
	102	+	if err := windows.Thread32Next(snap, &entry); err != nil {
	103	+	if errors.Is(err, windows.ERROR_NO_MORE_FILES) {
	104	+	break
	105	+	}
	106	+	return err
	107	+	}
	108	+	}
	109	+	if resumed == 0 {
	110	+	return errors.New("no thread to resume")
	111	+	}
	112	+	return nil
	113	+	}
	114	+
	115	+	// terminate kills every process in the job. Windows has no graceful group
	116	+	// signal, so force is ignored.
61	117		func (g *processGuard) terminate(_ bool) {
62	118		if g == nil {
63	119		return
64	120		}
65	121		g.mu.Lock()
66		-	job := g.job
67		-	g.mu.Unlock()
68		-	if job != 0 {
69		-	_ = windows.TerminateJobObject(job, 1)
70		-	return
71		-	}
72		-	if g.cmd != nil && g.cmd.Process != nil {
73		-	_ = g.cmd.Process.Kill()
	122	+	defer g.mu.Unlock()
	123	+	if g.job != 0 {
	124	+	_ = windows.TerminateJobObject(g.job, 1)
74	125		}
75	126		}
76	127
77		-	// alive is always false: terminate is already forceful.
78		-	func (g *processGuard) alive() bool { return false }
79		-
80		-	// release closes the job handle, which kills any survivor.
81		-	func (g *processGuard) release() {
	128	+	// cleanup kills the job and closes its handle, once.
	129	+	func (g *processGuard) cleanup() {
82	130		if g == nil {
83	131		return
84	132		}
85		-	g.mu.Lock()
86		-	job := g.job
87		-	g.job = 0
88		-	g.mu.Unlock()
89		-	if job != 0 {
90		-	_ = windows.CloseHandle(job)
91		-	}
	133	+	g.once.Do(func() {
	134	+	g.mu.Lock()
	135	+	defer g.mu.Unlock()
	136	+	if g.job != 0 {
	137	+	_ = windows.TerminateJobObject(g.job, 1)
	138	+	_ = windows.CloseHandle(g.job)
	139	+	g.job = 0
	140	+	}
	141	+	})
92	142		}


@@ -3,90 +3,140 @@
3		package mcp	3		package mcp
4			4
5		import (	5		import (
			6	+	"errors"
			7	+	"fmt"
6		"os/exec"	8		"os/exec"
7		"sync"	9		"sync"
			10	+	"syscall"
8		"unsafe"	11		"unsafe"
9			12
10		"golang.org/x/sys/windows"	13		"golang.org/x/sys/windows"
11		)	14		)
12			15
13	-	func isolateProcess(_ *exec.Cmd) {}	16	+	// isolateProcess starts the server suspended. The guard assigns it to a Job
			17	+	// Object before its first instruction runs and then resumes it, so no
			18	+	// descendant can be created outside the job.
			19	+	func isolateProcess(cmd *exec.Cmd) {
			20	+	if cmd.SysProcAttr == nil {
			21	+	cmd.SysProcAttr = &syscall.SysProcAttr{}
			22	+	}
			23	+	cmd.SysProcAttr.CreationFlags \|= windows.CREATE_SUSPENDED
			24	+	}
14			25
15	-	// processGuard assigns the server to a Job Object with	26	+	// processGuard owns a Job Object with JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE that
16	-	// JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, so descendants die with the server (and	27	+	// contains the server and all its descendants. The mutex is held across every
17	-	// with zut, should zut itself be killed).	28	+	// use of the handle so a concurrent release cannot close it mid-call.
18	-	//
19	-	// Assignment happens right after Start; a server that spawns children in the
20	-	// first instants of its life can escape the job. If the job cannot be created
21	-	// or assigned, the guard falls back to killing the direct child.
22		type processGuard struct {	29		type processGuard struct {
23	-	cmd *exec.Cmd	30	+	mu sync.Mutex
24	-		31	+	job windows.Handle // 0 once released
25	-	mu sync.Mutex	32	+	once sync.Once
26	-	job windows.Handle
27		}	33		}
28			34
29	-	func newProcessGuard(cmd exec.Cmd) processGuard {	35	+	// newProcessGuard is called right after cmd.Start with the process still
30	-	g := &processGuard{cmd: cmd}	36	+	// suspended. It fails closed: if the job cannot be set up, the child is killed
			37	+	// and an error is returned.
			38	+	func newProcessGuard(cmd exec.Cmd) (processGuard, error) {
31		if cmd == nil \|\| cmd.Process == nil {	39		if cmd == nil \|\| cmd.Process == nil {
32	-	return g	40	+	return nil, errors.New("process not started")
			41	+	}
			42	+	fail := func(what string, err error) (*processGuard, error) {
			43	+	_ = cmd.Process.Kill()
			44	+	return nil, fmt.Errorf("%s: %w", what, err)
33		}	45		}
34		job, err := windows.CreateJobObject(nil, nil)	46		job, err := windows.CreateJobObject(nil, nil)
35		if err != nil {	47		if err != nil {
36	-	return g	48	+	return fail("create job object", err)
37		}	49		}
38		var info windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION	50		var info windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION
39		info.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE	51		info.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
40		if _, err := windows.SetInformationJobObject(job, windows.JobObjectExtendedLimitInformation,	52		if _, err := windows.SetInformationJobObject(job, windows.JobObjectExtendedLimitInformation,
41		uintptr(unsafe.Pointer(&info)), uint32(unsafe.Sizeof(info))); err != nil {	53		uintptr(unsafe.Pointer(&info)), uint32(unsafe.Sizeof(info))); err != nil {
42		_ = windows.CloseHandle(job)	54		_ = windows.CloseHandle(job)
43	-	return g	55	+	return fail("configure job object", err)
44		}	56		}
45		proc, err := windows.OpenProcess(windows.PROCESS_SET_QUOTA\|windows.PROCESS_TERMINATE, false, uint32(cmd.Process.Pid))	57		proc, err := windows.OpenProcess(windows.PROCESS_SET_QUOTA\|windows.PROCESS_TERMINATE, false, uint32(cmd.Process.Pid))
46		if err != nil {	58		if err != nil {
47		_ = windows.CloseHandle(job)	59		_ = windows.CloseHandle(job)
48	-	return g	60	+	return fail("open server process", err)
49		}	61		}
50	-	defer windows.CloseHandle(proc)	62	+	err = windows.AssignProcessToJobObject(job, proc)
51	-	if err := windows.AssignProcessToJobObject(job, proc); err != nil {	63	+	_ = windows.CloseHandle(proc)
			64	+	if err != nil {
52		_ = windows.CloseHandle(job)	65		_ = windows.CloseHandle(job)
53	-	return g	66	+	return fail("assign server to job object", err)
54		}	67		}
55	-	g.job = job	68	+	if err := resumeProcess(uint32(cmd.Process.Pid)); err != nil {
56	-	return g	69	+	_ = windows.CloseHandle(job) // kills the job's only member
			70	+	return fail("resume server", err)
			71	+	}
			72	+	return &processGuard{job: job}, nil
57		}	73		}
58			74
59	-	// terminate kills the job (every process in it) or, without a job, the direct	75	+	// resumeProcess resumes every suspended thread of pid (a CREATE_SUSPENDED
60	-	// child. Windows has no graceful group signal, so force is ignored.	76	+	// process has exactly its main thread).
			77	+	func resumeProcess(pid uint32) error {
			78	+	snap, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPTHREAD, 0)
			79	+	if err != nil {
			80	+	return err
			81	+	}
			82	+	defer windows.CloseHandle(snap)
			83	+	entry := windows.ThreadEntry32{Size: uint32(unsafe.Sizeof(windows.ThreadEntry32{}))}
			84	+	if err := windows.Thread32First(snap, &entry); err != nil {
			85	+	return err
			86	+	}
			87	+	resumed := 0
			88	+	for {
			89	+	if entry.OwnerProcessID == pid {
			90	+	th, err := windows.OpenThread(windows.THREAD_SUSPEND_RESUME, false, entry.ThreadID)
			91	+	if err != nil {
			92	+	return err
			93	+	}
			94	+	_, err = windows.ResumeThread(th)
			95	+	_ = windows.CloseHandle(th)
			96	+	if err != nil {
			97	+	return err
			98	+	}
			99	+	resumed++
			100	+	}
			101	+	entry.Size = uint32(unsafe.Sizeof(entry))
			102	+	if err := windows.Thread32Next(snap, &entry); err != nil {
			103	+	if errors.Is(err, windows.ERROR_NO_MORE_FILES) {
			104	+	break
			105	+	}
			106	+	return err
			107	+	}
			108	+	}
			109	+	if resumed == 0 {
			110	+	return errors.New("no thread to resume")
			111	+	}
			112	+	return nil
			113	+	}
			114	+
			115	+	// terminate kills every process in the job. Windows has no graceful group
			116	+	// signal, so force is ignored.
61		func (g *processGuard) terminate(_ bool) {	117		func (g *processGuard) terminate(_ bool) {
62		if g == nil {	118		if g == nil {
63		return	119		return
64		}	120		}
65		g.mu.Lock()	121		g.mu.Lock()
66	-	job := g.job	122	+	defer g.mu.Unlock()
67	-	g.mu.Unlock()	123	+	if g.job != 0 {
68	-	if job != 0 {	124	+	_ = windows.TerminateJobObject(g.job, 1)
69	-	_ = windows.TerminateJobObject(job, 1)
70	-	return
71	-	}
72	-	if g.cmd != nil && g.cmd.Process != nil {
73	-	_ = g.cmd.Process.Kill()
74		}	125		}
75		}	126		}
76			127
77	-	// alive is always false: terminate is already forceful.	128	+	// cleanup kills the job and closes its handle, once.
78	-	func (g *processGuard) alive() bool { return false }	129	+	func (g *processGuard) cleanup() {
79	-
80	-	// release closes the job handle, which kills any survivor.
81	-	func (g *processGuard) release() {
82		if g == nil {	130		if g == nil {
83		return	131		return
84		}	132		}
85	-	g.mu.Lock()	133	+	g.once.Do(func() {
86	-	job := g.job	134	+	g.mu.Lock()
87	-	g.job = 0	135	+	defer g.mu.Unlock()
88	-	g.mu.Unlock()	136	+	if g.job != 0 {
89	-	if job != 0 {	137	+	_ = windows.TerminateJobObject(g.job, 1)
90	-	_ = windows.CloseHandle(job)	138	+	_ = windows.CloseHandle(g.job)
91	-	}	139	+	g.job = 0
			140	+	}
			141	+	})
92		}	142		}

A packages/agent/mcp/round2_test.go +275 -0

@@ -0,0 +1,275 @@
	1	+	package mcp
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"errors"
	7	+	"io"
	8	+	"net/url"
	9	+	"strings"
	10	+	"testing"
	11	+	"time"
	12	+	)
	13	+
	14	+	// Regression tests for the second review round.
	15	+
	16	+	// ---- GO-015: write cancellation ----
	17	+
	18	+	func TestStdioQueuedFrameIsWithdrawnButStartedWriteFailsTransport(t *testing.T) {
	19	+	c, events := connectFake(t, nil)
	20	+	st := c.t.(*stdioTransport)
	21	+
	22	+	// The server stops reading stdin.
	23	+	stallCtx, stallCancel := context.WithCancel(context.Background())
	24	+	defer stallCancel()
	25	+	go func() { _, _ = c.CallTool(stallCtx, "stall", []byte(`{}`)) }()
	26	+	waitEvent(t, events, "stalled")
	27	+
	28	+	// Frame A is far larger than the pipe buffer: its write starts and blocks.
	29	+	started := st.writesStarted.Load()
	30	+	aCtx, aCancel := context.WithCancel(context.Background())
	31	+	big := bytes.Repeat([]byte("a"), 4<<20)
	32	+	aDone := make(chan struct {
	33	+	sent bool
	34	+	err error
	35	+	}, 1)
	36	+	go func() {
	37	+	sent, err := st.writeFrame(aCtx, big)
	38	+	aDone <- struct {
	39	+	sent bool
	40	+	err error
	41	+	}{sent, err}
	42	+	}()
	43	+	eventually(t, "frame A to start writing", func() bool { return st.writesStarted.Load() == started+1 })
	44	+
	45	+	// Frame B queues behind A. Its caller gives up before B is dispatched:
	46	+	// B must be withdrawn, never written, and the transport must stay usable.
	47	+	bCtx, bCancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	48	+	defer bCancel()
	49	+	sent, err := st.writeFrame(bCtx, []byte(`{"jsonrpc":"2.0","method":"notifications/never"}`))
	50	+	if sent \|\| !errors.Is(err, context.DeadlineExceeded) {
	51	+	t.Fatalf("queued frame: sent=%v err=%v, want withdrawn with deadline error", sent, err)
	52	+	}
	53	+	select {
	54	+	case <-st.done():
	55	+	t.Fatal("withdrawing a queued frame must not fail the transport")
	56	+	default:
	57	+	}
	58	+
	59	+	// Now A's caller gives up mid-write: the frame cannot be withdrawn, so the
	60	+	// transport fails and stdin is closed.
	61	+	aCancel()
	62	+	select {
	63	+	case r := <-aDone:
	64	+	if !r.sent \|\| !errors.Is(r.err, context.Canceled) {
	65	+	t.Errorf("started frame: sent=%v err=%v", r.sent, r.err)
	66	+	}
	67	+	case <-time.After(5 * time.Second):
	68	+	t.Fatal("caller still blocked after cancellation")
	69	+	}
	70	+	select {
	71	+	case <-st.done():
	72	+	case <-time.After(5 * time.Second):
	73	+	t.Fatal("transport still live with a half-written frame")
	74	+	}
	75	+	if err := st.failure(); err == nil \|\| !strings.Contains(err.Error(), "interrupted") {
	76	+	t.Errorf("failure = %v", err)
	77	+	}
	78	+	// B was never handed to the writer.
	79	+	if got := st.writesStarted.Load(); got != started+1 {
	80	+	t.Errorf("writes started = %d, want %d (frame B must never be written)", got, started+1)
	81	+	}
	82	+	}
	83	+
	84	+	func TestManagerReconnectsAfterInterruptedWrite(t *testing.T) {
	85	+	m, events := startFake(t, nil, func(c *ServerConfig) { c.ToolTimeoutSec = 0.4 })
	86	+	// Park the server: the call is delivered, then it stops reading stdin.
	87	+	if res := execTool(t, findTool(t, m, "mcp__fake__stall"), `{}`); !res.IsError {
	88	+	t.Fatalf("stall result = %q", resultText(res))
	89	+	}
	90	+	waitEvent(t, events, "stalled")
	91	+	// A request too large for the pipe times out mid-write.
	92	+	big := `{"text":"` + strings.Repeat("a", 4<<20) + `"}`
	93	+	if res := execTool(t, findTool(t, m, "mcp__fake__echo"), big); !res.IsError {
	94	+	t.Fatalf("big call result = %q", resultText(res))
	95	+	}
	96	+	eventually(t, "status to show the dropped connection", func() bool { return !m.Status()[0].Connected })
	97	+	// The next call starts a fresh server and works.
	98	+	if got := resultText(execTool(t, findTool(t, m, "mcp__fake__echo"), `{"text":"again"}`)); got != "echo: again" {
	99	+	t.Fatalf("after reconnect = %q", got)
	100	+	}
	101	+	if n := countEvents(events, "start"); n != 2 {
	102	+	t.Errorf("server starts = %d, want 2", n)
	103	+	}
	104	+	}
	105	+
	106	+	// ---- GO-004, GO-018: stderr redaction ----
	107	+
	108	+	func newTestStderrTransport(cfg ServerConfig) *stdioTransport {
	109	+	return &stdioTransport{red: newRedactor(cfg), stderrDone: make(chan struct{})}
	110	+	}
	111	+
	112	+	func TestStderrSecretStraddlingTruncationIsRedacted(t *testing.T) {
	113	+	const secret = "SECRETVALUE-0123456789"
	114	+	tr := newTestStderrTransport(ServerConfig{Env: map[string]string{"K": secret}})
	115	+	// The secret starts before the 4096-byte display cut and ends after it.
	116	+	line := strings.Repeat("x", maxStderrLine-len(secret)/2) + secret + strings.Repeat("y", 300) + "\nnext line\n"
	117	+	tr.readStderr(strings.NewReader(line))
	118	+	tail := tr.stderrTail()
	119	+	for _, frag := range []string{"SECRET", "VALUE", "0123", secret[:8]} {
	120	+	if strings.Contains(tail, frag) {
	121	+	t.Errorf("secret fragment %q leaked: %.80q…", frag, tail[maxStderrLine-60:])
	122	+	}
	123	+	}
	124	+	if !strings.Contains(tail, redactedMarker) \|\| !strings.Contains(tail, "[line truncated]") \|\| !strings.Contains(tail, "next line") {
	125	+	t.Errorf("unexpected tail shape (len %d)", len(tail))
	126	+	}
	127	+	}
	128	+
	129	+	func TestStderrSecretOnTheCutIsNotPartiallyShown(t *testing.T) {
	130	+	// A line whose redacted form still ends in a secret prefix would leak it;
	131	+	// trimPartialSecret removes such a tail.
	132	+	red := newRedactor(ServerConfig{Env: map[string]string{"K": "TOPSECRET99"}})
	133	+	for _, in := range []string{"abc TOP", "abc TOPSECRET9", "abc TOPSECRET", "abc"} {
	134	+	got := red.trimPartialSecret(in)
	135	+	if strings.Contains(got, "TOP") {
	136	+	t.Errorf("trimPartialSecret(%q) = %q", in, got)
	137	+	}
	138	+	}
	139	+	if got := red.trimPartialSecret("keep TOPS and"); got != "keep TOPS and" {
	140	+	t.Errorf("interior text changed: %q", got)
	141	+	}
	142	+	var nilRed *redactor
	143	+	if got := nilRed.trimPartialSecret("x"); got != "x" {
	144	+	t.Error("nil redactor must pass text through")
	145	+	}
	146	+	}
	147	+
	148	+	func TestStdioStderrSecretEndToEnd(t *testing.T) {
	149	+	const secret = "E2ESECRET-abcdefghij"
	150	+	c, _ := connectFake(t, map[string]string{"FAKE_SECRET": secret})
	151	+	if got := callText(t, c, "stderr_secret", `{}`); got != "written" {
	152	+	t.Fatal(got)
	153	+	}
	154	+	st := c.t.(*stdioTransport)
	155	+	eventually(t, "stderr to be read", func() bool { return strings.Contains(st.stderrTail(), "after-secret") })
	156	+	tail := st.stderrTail()
	157	+	if strings.Contains(tail, "E2ESECRET") \|\| strings.Contains(tail, "abcdef") {
	158	+	t.Errorf("secret (or prefix) leaked")
	159	+	}
	160	+	}
	161	+
	162	+	func TestRedactionSurvivesControlSequencesInsideSecret(t *testing.T) {
	163	+	// Case 1: the server's output splits the secret with escape sequences.
	164	+	red := newRedactor(ServerConfig{Env: map[string]string{"K": "token-abcdef123"}})
	165	+	out := red.scrub("leak: tok\x1b[0men-abc\x1b]0;x\x07def123 end")
	166	+	if strings.Contains(out, "abcdef123") \|\| strings.Contains(out, "token") \|\| !strings.Contains(out, redactedMarker) {
	167	+	t.Errorf("out = %q", out)
	168	+	}
	169	+	// Case 2: the configured secret itself contains control bytes.
	170	+	red = newRedactor(ServerConfig{Env: map[string]string{"K": "se\x1b[1mcret-value-9"}})
	171	+	out = red.scrub("echo se\x1b[1mcret-value-9 / secret-value-9")
	172	+	if strings.Contains(out, "cret-value-9") {
	173	+	t.Errorf("out = %q", out)
	174	+	}
	175	+	}
	176	+
	177	+	// ---- GO-016: request-building errors ----
	178	+
	179	+	func TestHTTPRequestBuildErrorIsScrubbed(t *testing.T) {
	180	+	const secret = "QUERYSECRET-42"
	181	+	cfg := ServerConfig{Type: "http", URL: "http://bad host/path?token=" + secret}
	182	+	tr, err := newHTTPTransport("x", cfg, nil)
	183	+	if err != nil {
	184	+	t.Fatal(err)
	185	+	}
	186	+	defer tr.close()
	187	+	_, err = tr.request(context.Background(), 1, "initialize", nil)
	188	+	if err == nil {
	189	+	t.Fatal("want an error for an unparsable URL")
	190	+	}
	191	+	if strings.Contains(err.Error(), secret) {
	192	+	t.Errorf("error leaks the query token: %v", err)
	193	+	}
	194	+	// The same path also feeds close() and the GET stream.
	195	+	if _, err := tr.newRequest(context.Background(), "GET", nil); err == nil \|\| strings.Contains(err.Error(), secret) {
	196	+	t.Errorf("newRequest error = %v", err)
	197	+	}
	198	+	}
	199	+
	200	+	// ---- GO-017: encoded URL secrets ----
	201	+
	202	+	func TestURLSecretsIncludePercentDecodedForms(t *testing.T) {
	203	+	raw := "https://us%40er:p%40ssw0rd@h.example/p?api_key=a%2Bb%20c&other=plain-value&semi=x;y=z-token"
	204	+	got := map[string]bool{}
	205	+	for _, v := range urlSecrets(raw) {
	206	+	got[v] = true
	207	+	}
	208	+	for _, want := range []string{"us%40er", "us@er", "p%40ssw0rd", "p@ssw0rd", "a%2Bb%20c", "a+b c", "plain-value"} {
	209	+	if !got[want] {
	210	+	t.Errorf("urlSecrets missing %q (have %v)", want, got)
	211	+	}
	212	+	}
	213	+	red := newRedactor(ServerConfig{URL: raw})
	214	+	out := red.scrub("server said p@ssw0rd and a+b c and a%2Bb%20c and plain-value")
	215	+	for _, secret := range []string{"p@ssw0rd", "a+b c", "a%2Bb%20c", "plain-value"} {
	216	+	if strings.Contains(out, secret) {
	217	+	t.Errorf("%q leaked in %q", secret, out)
	218	+	}
	219	+	}
	220	+	// A URL that net/url refuses to parse still yields its secrets.
	221	+	unparsable := strings.Replace("http://bad_host/x?tok=SECRETQ99", "_", " ", 1) // space in the host
	222	+	if _, err := url.Parse(unparsable); err == nil {
	223	+	t.Fatal("test premise: URL should not parse")
	224	+	}
	225	+	bad := urlSecrets(unparsable)
	226	+	found := false
	227	+	for _, v := range bad {
	228	+	found = found \|\| v == "SECRETQ99"
	229	+	}
	230	+	if !found {
	231	+	t.Errorf("unparsable URL secrets = %v", bad)
	232	+	}
	233	+	}
	234	+
	235	+	// ---- GO-014: per-field budget ----
	236	+
	237	+	func TestBudgetedFieldIsBoundedBeforeFormatting(t *testing.T) {
	238	+	b := &budgetedText{}
	239	+	huge := strings.Repeat("é", 10<<20) // 20 MiB of two-byte runes
	240	+	got := b.field(huge)
	241	+	if len(got) > maxResultTextBytes+1 {
	242	+	t.Errorf("field not clipped: %d bytes", len(got))
	243	+	}
	244	+	if !strings.HasPrefix(huge, got) && got != "" {
	245	+	t.Error("clipped field is not a prefix")
	246	+	}
	247	+	if clipUTF8("日本語", 4) != "日" {
	248	+	t.Errorf("clipUTF8 split a rune: %q", clipUTF8("日本語", 4))
	249	+	}
	250	+	b.write(strings.Repeat("a", maxResultTextBytes-10))
	251	+	if got := b.field(strings.Repeat("b", 1000)); len(got) > 11 {
	252	+	t.Errorf("field ignores remaining budget: %d", len(got))
	253	+	}
	254	+	}
	255	+
	256	+	func TestListResourcesHugeDescription(t *testing.T) {
	257	+	m, _ := startFake(t, map[string]string{"FAKE_HUGE_DESC": "5242880"}, nil) // 5 MiB
	258	+	text := resultText(execTool(t, findTool(t, m, ListResourcesToolName), `{}`))
	259	+	if len(text) > maxResultTextBytes+len("\n... [truncated at 51200 bytes]") {
	260	+	t.Errorf("output %d bytes exceeds the budget", len(text))
	261	+	}
	262	+	if !strings.Contains(text, "mem://huge") \|\| !strings.Contains(text, "[truncated") {
	263	+	t.Errorf("unexpected output head: %.100q", text)
	264	+	}
	265	+	}
	266	+
	267	+	// ---- readStderr bookkeeping ----
	268	+
	269	+	func TestReadStderrHandlesEOFWithoutNewline(t *testing.T) {
	270	+	tr := newTestStderrTransport(ServerConfig{})
	271	+	tr.readStderr(io.MultiReader(strings.NewReader("one\ntwo"), strings.NewReader("")))
	272	+	if got := tr.stderrTail(); got != "one\ntwo" {
	273	+	t.Errorf("tail = %q", got)
	274	+	}
	275	+	}

@@ -0,0 +1,275 @@
	1	+	package mcp
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"errors"
	7	+	"io"
	8	+	"net/url"
	9	+	"strings"
	10	+	"testing"
	11	+	"time"
	12	+	)
	13	+
	14	+	// Regression tests for the second review round.
	15	+
	16	+	// ---- GO-015: write cancellation ----
	17	+
	18	+	func TestStdioQueuedFrameIsWithdrawnButStartedWriteFailsTransport(t *testing.T) {
	19	+	c, events := connectFake(t, nil)
	20	+	st := c.t.(*stdioTransport)
	21	+
	22	+	// The server stops reading stdin.
	23	+	stallCtx, stallCancel := context.WithCancel(context.Background())
	24	+	defer stallCancel()
	25	+	go func() { _, _ = c.CallTool(stallCtx, "stall", []byte(`{}`)) }()
	26	+	waitEvent(t, events, "stalled")
	27	+
	28	+	// Frame A is far larger than the pipe buffer: its write starts and blocks.
	29	+	started := st.writesStarted.Load()
	30	+	aCtx, aCancel := context.WithCancel(context.Background())
	31	+	big := bytes.Repeat([]byte("a"), 4<<20)
	32	+	aDone := make(chan struct {
	33	+	sent bool
	34	+	err error
	35	+	}, 1)
	36	+	go func() {
	37	+	sent, err := st.writeFrame(aCtx, big)
	38	+	aDone <- struct {
	39	+	sent bool
	40	+	err error
	41	+	}{sent, err}
	42	+	}()
	43	+	eventually(t, "frame A to start writing", func() bool { return st.writesStarted.Load() == started+1 })
	44	+
	45	+	// Frame B queues behind A. Its caller gives up before B is dispatched:
	46	+	// B must be withdrawn, never written, and the transport must stay usable.
	47	+	bCtx, bCancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	48	+	defer bCancel()
	49	+	sent, err := st.writeFrame(bCtx, []byte(`{"jsonrpc":"2.0","method":"notifications/never"}`))
	50	+	if sent \|\| !errors.Is(err, context.DeadlineExceeded) {
	51	+	t.Fatalf("queued frame: sent=%v err=%v, want withdrawn with deadline error", sent, err)
	52	+	}
	53	+	select {
	54	+	case <-st.done():
	55	+	t.Fatal("withdrawing a queued frame must not fail the transport")
	56	+	default:
	57	+	}
	58	+
	59	+	// Now A's caller gives up mid-write: the frame cannot be withdrawn, so the
	60	+	// transport fails and stdin is closed.
	61	+	aCancel()
	62	+	select {
	63	+	case r := <-aDone:
	64	+	if !r.sent \|\| !errors.Is(r.err, context.Canceled) {
	65	+	t.Errorf("started frame: sent=%v err=%v", r.sent, r.err)
	66	+	}
	67	+	case <-time.After(5 * time.Second):
	68	+	t.Fatal("caller still blocked after cancellation")
	69	+	}
	70	+	select {
	71	+	case <-st.done():
	72	+	case <-time.After(5 * time.Second):
	73	+	t.Fatal("transport still live with a half-written frame")
	74	+	}
	75	+	if err := st.failure(); err == nil \|\| !strings.Contains(err.Error(), "interrupted") {
	76	+	t.Errorf("failure = %v", err)
	77	+	}
	78	+	// B was never handed to the writer.
	79	+	if got := st.writesStarted.Load(); got != started+1 {
	80	+	t.Errorf("writes started = %d, want %d (frame B must never be written)", got, started+1)
	81	+	}
	82	+	}
	83	+
	84	+	func TestManagerReconnectsAfterInterruptedWrite(t *testing.T) {
	85	+	m, events := startFake(t, nil, func(c *ServerConfig) { c.ToolTimeoutSec = 0.4 })
	86	+	// Park the server: the call is delivered, then it stops reading stdin.
	87	+	if res := execTool(t, findTool(t, m, "mcp__fake__stall"), `{}`); !res.IsError {
	88	+	t.Fatalf("stall result = %q", resultText(res))
	89	+	}
	90	+	waitEvent(t, events, "stalled")
	91	+	// A request too large for the pipe times out mid-write.
	92	+	big := `{"text":"` + strings.Repeat("a", 4<<20) + `"}`
	93	+	if res := execTool(t, findTool(t, m, "mcp__fake__echo"), big); !res.IsError {
	94	+	t.Fatalf("big call result = %q", resultText(res))
	95	+	}
	96	+	eventually(t, "status to show the dropped connection", func() bool { return !m.Status()[0].Connected })
	97	+	// The next call starts a fresh server and works.
	98	+	if got := resultText(execTool(t, findTool(t, m, "mcp__fake__echo"), `{"text":"again"}`)); got != "echo: again" {
	99	+	t.Fatalf("after reconnect = %q", got)
	100	+	}
	101	+	if n := countEvents(events, "start"); n != 2 {
	102	+	t.Errorf("server starts = %d, want 2", n)
	103	+	}
	104	+	}
	105	+
	106	+	// ---- GO-004, GO-018: stderr redaction ----
	107	+
	108	+	func newTestStderrTransport(cfg ServerConfig) *stdioTransport {
	109	+	return &stdioTransport{red: newRedactor(cfg), stderrDone: make(chan struct{})}
	110	+	}
	111	+
	112	+	func TestStderrSecretStraddlingTruncationIsRedacted(t *testing.T) {
	113	+	const secret = "SECRETVALUE-0123456789"
	114	+	tr := newTestStderrTransport(ServerConfig{Env: map[string]string{"K": secret}})
	115	+	// The secret starts before the 4096-byte display cut and ends after it.
	116	+	line := strings.Repeat("x", maxStderrLine-len(secret)/2) + secret + strings.Repeat("y", 300) + "\nnext line\n"
	117	+	tr.readStderr(strings.NewReader(line))
	118	+	tail := tr.stderrTail()
	119	+	for _, frag := range []string{"SECRET", "VALUE", "0123", secret[:8]} {
	120	+	if strings.Contains(tail, frag) {
	121	+	t.Errorf("secret fragment %q leaked: %.80q…", frag, tail[maxStderrLine-60:])
	122	+	}
	123	+	}
	124	+	if !strings.Contains(tail, redactedMarker) \|\| !strings.Contains(tail, "[line truncated]") \|\| !strings.Contains(tail, "next line") {
	125	+	t.Errorf("unexpected tail shape (len %d)", len(tail))
	126	+	}
	127	+	}
	128	+
	129	+	func TestStderrSecretOnTheCutIsNotPartiallyShown(t *testing.T) {
	130	+	// A line whose redacted form still ends in a secret prefix would leak it;
	131	+	// trimPartialSecret removes such a tail.
	132	+	red := newRedactor(ServerConfig{Env: map[string]string{"K": "TOPSECRET99"}})
	133	+	for _, in := range []string{"abc TOP", "abc TOPSECRET9", "abc TOPSECRET", "abc"} {
	134	+	got := red.trimPartialSecret(in)
	135	+	if strings.Contains(got, "TOP") {
	136	+	t.Errorf("trimPartialSecret(%q) = %q", in, got)
	137	+	}
	138	+	}
	139	+	if got := red.trimPartialSecret("keep TOPS and"); got != "keep TOPS and" {
	140	+	t.Errorf("interior text changed: %q", got)
	141	+	}
	142	+	var nilRed *redactor
	143	+	if got := nilRed.trimPartialSecret("x"); got != "x" {
	144	+	t.Error("nil redactor must pass text through")
	145	+	}
	146	+	}
	147	+
	148	+	func TestStdioStderrSecretEndToEnd(t *testing.T) {
	149	+	const secret = "E2ESECRET-abcdefghij"
	150	+	c, _ := connectFake(t, map[string]string{"FAKE_SECRET": secret})
	151	+	if got := callText(t, c, "stderr_secret", `{}`); got != "written" {
	152	+	t.Fatal(got)
	153	+	}
	154	+	st := c.t.(*stdioTransport)
	155	+	eventually(t, "stderr to be read", func() bool { return strings.Contains(st.stderrTail(), "after-secret") })
	156	+	tail := st.stderrTail()
	157	+	if strings.Contains(tail, "E2ESECRET") \|\| strings.Contains(tail, "abcdef") {
	158	+	t.Errorf("secret (or prefix) leaked")
	159	+	}
	160	+	}
	161	+
	162	+	func TestRedactionSurvivesControlSequencesInsideSecret(t *testing.T) {
	163	+	// Case 1: the server's output splits the secret with escape sequences.
	164	+	red := newRedactor(ServerConfig{Env: map[string]string{"K": "token-abcdef123"}})
	165	+	out := red.scrub("leak: tok\x1b[0men-abc\x1b]0;x\x07def123 end")
	166	+	if strings.Contains(out, "abcdef123") \|\| strings.Contains(out, "token") \|\| !strings.Contains(out, redactedMarker) {
	167	+	t.Errorf("out = %q", out)
	168	+	}
	169	+	// Case 2: the configured secret itself contains control bytes.
	170	+	red = newRedactor(ServerConfig{Env: map[string]string{"K": "se\x1b[1mcret-value-9"}})
	171	+	out = red.scrub("echo se\x1b[1mcret-value-9 / secret-value-9")
	172	+	if strings.Contains(out, "cret-value-9") {
	173	+	t.Errorf("out = %q", out)
	174	+	}
	175	+	}
	176	+
	177	+	// ---- GO-016: request-building errors ----
	178	+
	179	+	func TestHTTPRequestBuildErrorIsScrubbed(t *testing.T) {
	180	+	const secret = "QUERYSECRET-42"
	181	+	cfg := ServerConfig{Type: "http", URL: "http://bad host/path?token=" + secret}
	182	+	tr, err := newHTTPTransport("x", cfg, nil)
	183	+	if err != nil {
	184	+	t.Fatal(err)
	185	+	}
	186	+	defer tr.close()
	187	+	_, err = tr.request(context.Background(), 1, "initialize", nil)
	188	+	if err == nil {
	189	+	t.Fatal("want an error for an unparsable URL")
	190	+	}
	191	+	if strings.Contains(err.Error(), secret) {
	192	+	t.Errorf("error leaks the query token: %v", err)
	193	+	}
	194	+	// The same path also feeds close() and the GET stream.
	195	+	if _, err := tr.newRequest(context.Background(), "GET", nil); err == nil \|\| strings.Contains(err.Error(), secret) {
	196	+	t.Errorf("newRequest error = %v", err)
	197	+	}
	198	+	}
	199	+
	200	+	// ---- GO-017: encoded URL secrets ----
	201	+
	202	+	func TestURLSecretsIncludePercentDecodedForms(t *testing.T) {
	203	+	raw := "https://us%40er:p%40ssw0rd@h.example/p?api_key=a%2Bb%20c&other=plain-value&semi=x;y=z-token"
	204	+	got := map[string]bool{}
	205	+	for _, v := range urlSecrets(raw) {
	206	+	got[v] = true
	207	+	}
	208	+	for _, want := range []string{"us%40er", "us@er", "p%40ssw0rd", "p@ssw0rd", "a%2Bb%20c", "a+b c", "plain-value"} {
	209	+	if !got[want] {
	210	+	t.Errorf("urlSecrets missing %q (have %v)", want, got)
	211	+	}
	212	+	}
	213	+	red := newRedactor(ServerConfig{URL: raw})
	214	+	out := red.scrub("server said p@ssw0rd and a+b c and a%2Bb%20c and plain-value")
	215	+	for _, secret := range []string{"p@ssw0rd", "a+b c", "a%2Bb%20c", "plain-value"} {
	216	+	if strings.Contains(out, secret) {
	217	+	t.Errorf("%q leaked in %q", secret, out)
	218	+	}
	219	+	}
	220	+	// A URL that net/url refuses to parse still yields its secrets.
	221	+	unparsable := strings.Replace("http://bad_host/x?tok=SECRETQ99", "_", " ", 1) // space in the host
	222	+	if _, err := url.Parse(unparsable); err == nil {
	223	+	t.Fatal("test premise: URL should not parse")
	224	+	}
	225	+	bad := urlSecrets(unparsable)
	226	+	found := false
	227	+	for _, v := range bad {
	228	+	found = found \|\| v == "SECRETQ99"
	229	+	}
	230	+	if !found {
	231	+	t.Errorf("unparsable URL secrets = %v", bad)
	232	+	}
	233	+	}
	234	+
	235	+	// ---- GO-014: per-field budget ----
	236	+
	237	+	func TestBudgetedFieldIsBoundedBeforeFormatting(t *testing.T) {
	238	+	b := &budgetedText{}
	239	+	huge := strings.Repeat("é", 10<<20) // 20 MiB of two-byte runes
	240	+	got := b.field(huge)
	241	+	if len(got) > maxResultTextBytes+1 {
	242	+	t.Errorf("field not clipped: %d bytes", len(got))
	243	+	}
	244	+	if !strings.HasPrefix(huge, got) && got != "" {
	245	+	t.Error("clipped field is not a prefix")
	246	+	}
	247	+	if clipUTF8("日本語", 4) != "日" {
	248	+	t.Errorf("clipUTF8 split a rune: %q", clipUTF8("日本語", 4))
	249	+	}
	250	+	b.write(strings.Repeat("a", maxResultTextBytes-10))
	251	+	if got := b.field(strings.Repeat("b", 1000)); len(got) > 11 {
	252	+	t.Errorf("field ignores remaining budget: %d", len(got))
	253	+	}
	254	+	}
	255	+
	256	+	func TestListResourcesHugeDescription(t *testing.T) {
	257	+	m, _ := startFake(t, map[string]string{"FAKE_HUGE_DESC": "5242880"}, nil) // 5 MiB
	258	+	text := resultText(execTool(t, findTool(t, m, ListResourcesToolName), `{}`))
	259	+	if len(text) > maxResultTextBytes+len("\n... [truncated at 51200 bytes]") {
	260	+	t.Errorf("output %d bytes exceeds the budget", len(text))
	261	+	}
	262	+	if !strings.Contains(text, "mem://huge") \|\| !strings.Contains(text, "[truncated") {
	263	+	t.Errorf("unexpected output head: %.100q", text)
	264	+	}
	265	+	}
	266	+
	267	+	// ---- readStderr bookkeeping ----
	268	+
	269	+	func TestReadStderrHandlesEOFWithoutNewline(t *testing.T) {
	270	+	tr := newTestStderrTransport(ServerConfig{})
	271	+	tr.readStderr(io.MultiReader(strings.NewReader("one\ntwo"), strings.NewReader("")))
	272	+	if got := tr.stderrTail(); got != "one\ntwo" {
	273	+	t.Errorf("tail = %q", got)
	274	+	}
	275	+	}

M packages/agent/mcp/sanitize.go +75 -12

@@ -1,6 +1,7 @@
1	1		package mcp
2	2
3	3		import (
	4	+	"net/url"
4	5		"os"
5	6		"regexp"
6	7		"sort"

@@ -24,6 +25,9 @@ const (
24	25		// A nil *redactor still strips URLs' credentials and control characters.
25	26		type redactor struct {
26	27		secrets []string
	28	+	// maxLen is the length of the longest secret, used to size the overlap
	29	+	// when long lines are cut for display.
	30	+	maxLen int
27	31		}
28	32
29	33		// newRedactor collects the secret-bearing values of one server configuration:

@@ -37,6 +41,12 @@ func newRedactor(cfg ServerConfig) *redactor {
37	41		return
38	42		}
39	43		set[v] = true
	44	+	// The control-stripped form too: text is sanitized before it is
	45	+	// matched, so a secret that itself contains escape bytes must match
	46	+	// in its sanitized shape.
	47	+	if clean := sanitizeText(v, true); len(clean) >= minSecretLen {
	48	+	set[clean] = true
	49	+	}
40	50		// "Bearer abc123" is also worth redacting as just "abc123".
41	51		if strings.ContainsAny(v, " \t") {
42	52		for _, field := range strings.Fields(v) {

@@ -75,6 +85,7 @@ func newRedactor(cfg ServerConfig) *redactor {
75	85		r := &redactor{}
76	86		for v := range set {
77	87		r.secrets = append(r.secrets, v)
	88	+	r.maxLen = max(r.maxLen, len(v))
78	89		}
79	90		// Longest first so a secret containing another is replaced whole.
80	91		sort.Slice(r.secrets, func(i, j int) bool {

@@ -86,30 +97,52 @@ func newRedactor(cfg ServerConfig) *redactor {
86	97		return r
87	98		}
88	99
89		-	// urlSecrets returns the userinfo password/user and query values of raw.
	100	+	// urlSecrets returns the secret-bearing parts of raw: userinfo (user and
	101	+	// password) and query values, both as written and percent-decoded. It splits
	102	+	// the text itself instead of requiring url.Parse to succeed, because the URLs
	103	+	// that matter most here are the malformed ones that fail to parse.
90	104		func urlSecrets(raw string) []string {
	105	+	var out []string
	106	+	add := func(v string) {
	107	+	if v == "" {
	108	+	return
	109	+	}
	110	+	out = append(out, v)
	111	+	if d, err := url.PathUnescape(v); err == nil && d != v {
	112	+	out = append(out, d)
	113	+	}
	114	+	if d, err := url.QueryUnescape(v); err == nil && d != v {
	115	+	out = append(out, d)
	116	+	}
	117	+	}
91	118		rest := raw
92	119		if i := strings.Index(rest, "://"); i >= 0 {
93	120		rest = rest[i+3:]
94	121		}
95		-	var out []string
96	122		authority := rest
97	123		if i := strings.IndexAny(authority, "/?#"); i >= 0 {
98	124		authority = authority[:i]
99	125		}
100	126		if i := strings.LastIndex(authority, "@"); i >= 0 {
101		-	for _, part := range strings.SplitN(authority[:i], ":", 2) {
102		-	out = append(out, part)
103		-	}
	127	+	user, pass, _ := strings.Cut(authority[:i], ":")
	128	+	add(user)
	129	+	add(pass)
104	130		}
105	131		if i := strings.Index(rest, "?"); i >= 0 {
106	132		query := rest[i+1:]
107	133		if j := strings.Index(query, "#"); j >= 0 {
108	134		query = query[:j]
109	135		}
110		-	for _, pair := range strings.Split(query, "&") {
	136	+	for _, pair := range strings.FieldsFunc(query, func(r rune) bool { return r == '&' \|\| r == ';' }) {
111	137		if _, v, ok := strings.Cut(pair, "="); ok {
112		-	out = append(out, v)
	138	+	add(v)
	139	+	}
	140	+	}
	141	+	if values, err := url.ParseQuery(query); err == nil {
	142	+	for _, vs := range values {
	143	+	for _, v := range vs {
	144	+	add(v)
	145	+	}
113	146		}
114	147		}
115	148		}

@@ -122,13 +155,43 @@ func (r *redactor) scrub(s string) string {
122	155		if s == "" {
123	156		return s
124	157		}
125		-	if r != nil {
126		-	for _, secret := range r.secrets {
127		-	s = strings.ReplaceAll(s, secret, redactedMarker)
	158	+	// Redact, strip control sequences, then redact again: a secret that has
	159	+	// escape bytes inserted into it only becomes matchable once they are gone.
	160	+	s = r.redact(s)
	161	+	s = sanitizeText(s, true)
	162	+	s = r.redact(s)
	163	+	return scrubURLs(s)
	164	+	}
	165	+
	166	+	// redact replaces configured secret values. A nil redactor changes nothing.
	167	+	func (r *redactor) redact(s string) string {
	168	+	if r == nil {
	169	+	return s
	170	+	}
	171	+	for _, secret := range r.secrets {
	172	+	s = strings.ReplaceAll(s, secret, redactedMarker)
	173	+	}
	174	+	return s
	175	+	}
	176	+
	177	+	// trimPartialSecret removes a trailing fragment that is a proper prefix of a
	178	+	// secret. It is for text that was cut at an arbitrary byte: a secret that
	179	+	// straddles the cut cannot be matched as a whole, and its prefix must not
	180	+	// survive either.
	181	+	func (r *redactor) trimPartialSecret(s string) string {
	182	+	if r == nil {
	183	+	return s
	184	+	}
	185	+	drop := 0
	186	+	for _, secret := range r.secrets {
	187	+	for k := min(len(secret)-1, len(s)); k > drop; k-- {
	188	+	if strings.HasSuffix(s, secret[:k]) {
	189	+	drop = k
	190	+	break
	191	+	}
128	192		}
129	193		}
130		-	s = scrubURLs(s)
131		-	return sanitizeText(s, true)
	194	+	return s[:len(s)-drop]
132	195		}
133	196
134	197		// scrubOneLine is scrub with newlines collapsed to spaces.

 package mcp
 import (
+	"net/url"
 	"os"
 	"regexp"
 	"sort"
 // A nil *redactor still strips URLs' credentials and control characters.
 type redactor struct {
 	secrets []string
+	// maxLen is the length of the longest secret, used to size the overlap
+	// when long lines are cut for display.
+	maxLen int
 }
 // newRedactor collects the secret-bearing values of one server configuration:
 			return
 		}
 		set[v] = true
+		// The control-stripped form too: text is sanitized before it is
+		// matched, so a secret that itself contains escape bytes must match
+		// in its sanitized shape.
+		if clean := sanitizeText(v, true); len(clean) >= minSecretLen {
+			set[clean] = true
+		}
 		// "Bearer abc123" is also worth redacting as just "abc123".
 		if strings.ContainsAny(v, " \t") {
 			for _, field := range strings.Fields(v) {
 	r := &redactor{}
 	for v := range set {
 		r.secrets = append(r.secrets, v)
+		r.maxLen = max(r.maxLen, len(v))
 	}
 	// Longest first so a secret containing another is replaced whole.
 	sort.Slice(r.secrets, func(i, j int) bool {
 	return r
 }
-// urlSecrets returns the userinfo password/user and query values of raw.
+// urlSecrets returns the secret-bearing parts of raw: userinfo (user and
+// password) and query values, both as written and percent-decoded. It splits
+// the text itself instead of requiring url.Parse to succeed, because the URLs
+// that matter most here are the malformed ones that fail to parse.
 func urlSecrets(raw string) []string {
+	var out []string
+	add := func(v string) {
+		if v == "" {
+			return
+		}
+		out = append(out, v)
+		if d, err := url.PathUnescape(v); err == nil && d != v {
+			out = append(out, d)
+		}
+		if d, err := url.QueryUnescape(v); err == nil && d != v {
+			out = append(out, d)
+		}
+	}
 	rest := raw
 	if i := strings.Index(rest, "://"); i >= 0 {
 		rest = rest[i+3:]
 	}
-	var out []string
 	authority := rest
 	if i := strings.IndexAny(authority, "/?#"); i >= 0 {
 		authority = authority[:i]
 	}
 	if i := strings.LastIndex(authority, "@"); i >= 0 {
-		for _, part := range strings.SplitN(authority[:i], ":", 2) {
+		user, pass, _ := strings.Cut(authority[:i], ":")
-			out = append(out, part)
+		add(user)
-		}
+		add(pass)
 	}
 	if i := strings.Index(rest, "?"); i >= 0 {
 		query := rest[i+1:]
 		if j := strings.Index(query, "#"); j >= 0 {
 			query = query[:j]
 		}
-		for _, pair := range strings.Split(query, "&") {
+		for _, pair := range strings.FieldsFunc(query, func(r rune) bool { return r == '&' || r == ';' }) {
 			if _, v, ok := strings.Cut(pair, "="); ok {
-				out = append(out, v)
+				add(v)
+			}
+		}
+		if values, err := url.ParseQuery(query); err == nil {
+			for _, vs := range values {
+				for _, v := range vs {
+					add(v)
+				}
 			}
 		}
 	}
 	if s == "" {
 		return s
 	}
-	if r != nil {
+	// Redact, strip control sequences, then redact again: a secret that has
-		for _, secret := range r.secrets {
+	// escape bytes inserted into it only becomes matchable once they are gone.
-			s = strings.ReplaceAll(s, secret, redactedMarker)
+	s = r.redact(s)
+	s = sanitizeText(s, true)
+	s = r.redact(s)
+	return scrubURLs(s)
+}
+// redact replaces configured secret values. A nil redactor changes nothing.
+func (r *redactor) redact(s string) string {
+	if r == nil {
+		return s
+	}
+	for _, secret := range r.secrets {
+		s = strings.ReplaceAll(s, secret, redactedMarker)
+	}
+	return s
+}
+// trimPartialSecret removes a trailing fragment that is a proper prefix of a
+// secret. It is for text that was cut at an arbitrary byte: a secret that
+// straddles the cut cannot be matched as a whole, and its prefix must not
+// survive either.
+func (r *redactor) trimPartialSecret(s string) string {
+	if r == nil {
+		return s
+	}
+	drop := 0
+	for _, secret := range r.secrets {
+		for k := min(len(secret)-1, len(s)); k > drop; k-- {
+			if strings.HasSuffix(s, secret[:k]) {
+				drop = k
+				break
+			}
 		}
 	}
-	s = scrubURLs(s)
+	return s[:len(s)-drop]
-	return sanitizeText(s, true)
 }
 // scrubOneLine is scrub with newlines collapsed to spaces.

M packages/agent/mcp/stdio.go +88 -34


@@ -15,6 +15,7 @@ import (
15	15		"sync"
16	16		"sync/atomic"
17	17		"time"
	18	+	"unicode/utf8"
18	19		)
19	20
20	21		// maxStdioFrame bounds one newline-delimited JSON-RPC message from a server.

@@ -26,6 +27,9 @@ const (
26	27		// maxStderrLine truncates one retained stderr line; the rest of a longer
27	28		// line is still drained from the pipe, just not kept.
28	29		maxStderrLine = 4096
	30	+	// maxStderrOverlap caps how far past maxStderrLine raw bytes are kept so
	31	+	// a straddling secret can still be redacted.
	32	+	maxStderrOverlap = 64 << 10
29	33		// writeQueueSize bounds frames waiting for the writer goroutine.
30	34		writeQueueSize = 64
31	35		// responseWriteTimeout bounds answering one server-initiated request.

@@ -57,8 +61,17 @@ type writeReq struct {
57	61		ctx context.Context
58	62		frame []byte
59	63		res chan error
	64	+	state atomic.Int32 // reqQueued, reqWriting, or reqSkipped
60	65		}
61	66
	67	+	// A queued frame is claimed exactly once: by the writer (it will be written)
	68	+	// or by its caller (it will never be written).
	69	+	const (
	70	+	reqQueued int32 = iota
	71	+	reqWriting
	72	+	reqSkipped
	73	+	)
	74	+
62	75		type stdioTransport struct {
63	76		name string
64	77		cmd *exec.Cmd

@@ -92,6 +105,9 @@ type stdioTransport struct {
92	105		// cancelsSent counts notifications/cancelled sends started (diagnostics
93	106		// and tests).
94	107		cancelsSent atomic.Int64
	108	+	// writesStarted counts frames the writer has begun writing (diagnostics
	109	+	// and tests).
	110	+	writesStarted atomic.Int64
95	111		}
96	112
97	113		// buildStdioEnv returns the restricted child environment.

@@ -153,11 +169,21 @@ func startStdio(name string, cfg ServerConfig, cwd string) (*stdioTransport, err
153	169		}
154	170		outW.Close()
155	171		errW.Close()
	172	+	guard, err := newProcessGuard(cmd)
	173	+	if err != nil {
	174	+	// Fail closed: never run a server that cannot be contained.
	175	+	_ = stdin.Close()
	176	+	_ = cmd.Process.Kill()
	177	+	_ = cmd.Wait()
	178	+	outR.Close()
	179	+	errR.Close()
	180	+	return nil, fmt.Errorf("start %s: %w", cfg.Command, red.scrubError(err))
	181	+	}
156	182		ctx, cancel := context.WithCancel(context.Background())
157	183		t := &stdioTransport{
158	184		name: name,
159	185		cmd: cmd,
160		-	guard: newProcessGuard(cmd),
	186	+	guard: guard,
161	187		red: red,
162	188		ctx: ctx,
163	189		cancel: cancel,

@@ -187,6 +213,9 @@ func startStdio(name string, cfg ServerConfig, cwd string) (*stdioTransport, err
187	213		case <-time.After(250 * time.Millisecond):
188	214		}
189	215		t.fail(fmt.Errorf("server process ended: %w", waitErr))
	216	+	// The leader is gone: stop whatever it left behind, once, while the
	217	+	// process-group id is still guaranteed to belong to our group.
	218	+	t.guard.cleanup()
190	219		}()
191	220		return t, nil
192	221		}

@@ -194,15 +223,28 @@ func startStdio(name string, cfg ServerConfig, cwd string) (*stdioTransport, err
194	223		// readStderr drains the child's stderr for its whole life, keeping a bounded
195	224		// tail of bounded lines. It never stops early: a child blocked on a full
196	225		// stderr pipe would hang.
	226	+	//
	227	+	// Redaction happens on the complete line before it is shortened for display.
	228	+	// For a line longer than the display limit the raw prefix is kept
	229	+	// maxSecretLen bytes past the cut, so a secret that straddles the cut is
	230	+	// matched whole; what remains of one anyway is trimmed afterwards.
197	231		func (t *stdioTransport) readStderr(r io.Reader) {
198	232		defer close(t.stderrDone)
199	233		reader := bufio.NewReaderSize(r, 4096)
	234	+	keep := maxStderrLine + min(t.red.maxLen, maxStderrOverlap)
200	235		var line []byte
201	236		truncated := false
202	237		flush := func() {
203		-	text := strings.TrimRight(string(line), "\r")
204		-	if truncated {
205		-	text += "…[line truncated]"
	238	+	text := t.red.scrub(strings.TrimRight(string(line), "\r"))
	239	+	if len(text) > maxStderrLine \|\| truncated {
	240	+	if len(text) > maxStderrLine {
	241	+	cut := maxStderrLine
	242	+	for cut > 0 && !utf8.RuneStart(text[cut]) {
	243	+	cut--
	244	+	}
	245	+	text = text[:cut]
	246	+	}
	247	+	text = t.red.trimPartialSecret(text) + "…[line truncated]"
206	248		}
207	249		line, truncated = nil, false
208	250		t.stderrMu.Lock()

@@ -214,7 +256,7 @@ func (t *stdioTransport) readStderr(r io.Reader) {
214	256		}
215	257		for {
216	258		chunk, isPrefix, err := reader.ReadLine()
217		-	if room := maxStderrLine - len(line); len(chunk) > room {
	259	+	if room := keep - len(line); len(chunk) > room {
218	260		if room > 0 {
219	261		line = append(line, chunk[:room]...)
220	262		}

@@ -335,10 +377,14 @@ func (t *stdioTransport) writeLoop() {
335	377		case <-t.doneCh:
336	378		return
337	379		case req := <-t.writeQ:
338		-	if err := req.ctx.Err(); err != nil {
	380	+	if err := req.ctx.Err(); err != nil && req.state.CompareAndSwap(reqQueued, reqSkipped) {
339	381		req.res <- err // never started: nothing partial on the wire
340	382		continue
341	383		}
	384	+	if !req.state.CompareAndSwap(reqQueued, reqWriting) {
	385	+	continue // the caller gave up before we started
	386	+	}
	387	+	t.writesStarted.Add(1)
342	388		_, err := t.stdin.Write(req.frame)
343	389		req.res <- err
344	390		if err != nil {

@@ -353,26 +399,49 @@ func (t *stdioTransport) writeLoop() {
353	399		// write queues one frame (newline appended) and waits until it has been
354	400		// written, ctx ends, or the connection fails.
355	401		func (t *stdioTransport) write(ctx context.Context, frame []byte) error {
	402	+	_, err := t.writeFrame(ctx, frame)
	403	+	return err
	404	+	}
	405	+
	406	+	// writeFrame is write that also reports whether any byte of the frame may
	407	+	// have reached the server. When ctx ends:
	408	+	// - a frame the writer has not started is withdrawn and never written, and
	409	+	// the transport stays usable;
	410	+	// - a frame whose write is already in progress cannot be withdrawn (part of
	411	+	// it may be on the wire and the rest could still be flushed later), so the
	412	+	// transport is failed and stdin closed. The manager reconnects on the next
	413	+	// call and the abandoned frame can never complete.
	414	+	func (t *stdioTransport) writeFrame(ctx context.Context, frame []byte) (sent bool, err error) {
356	415		req := &writeReq{ctx: ctx, frame: append(append(make([]byte, 0, len(frame)+1), frame...), '\n'), res: make(chan error, 1)}
357	416		select {
358	417		case t.writeQ <- req:
359	418		case <-ctx.Done():
360		-	return ctx.Err()
	419	+	return false, ctx.Err()
361	420		case <-t.doneCh:
362		-	return t.failure()
	421	+	return false, t.failure()
363	422		}
364	423		select {
365	424		case err := <-req.res:
366		-	return err
	425	+	return req.state.Load() == reqWriting, err
367	426		case <-ctx.Done():
368		-	return ctx.Err()
	427	+	if req.state.CompareAndSwap(reqQueued, reqSkipped) {
	428	+	return false, ctx.Err()
	429	+	}
	430	+	select {
	431	+	case err := <-req.res:
	432	+	return req.state.Load() == reqWriting, err
	433	+	default:
	434	+	}
	435	+	t.fail(fmt.Errorf("write to server interrupted: %w", ctx.Err()))
	436	+	_ = t.stdin.Close() // unblocks the stuck write
	437	+	return true, ctx.Err()
369	438		case <-t.doneCh:
370	439		select {
371	440		case err := <-req.res:
372		-	return err
	441	+	return req.state.Load() == reqWriting, err
373	442		default:
374	443		}
375		-	return t.failure()
	444	+	return req.state.Load() == reqWriting, t.failure()
376	445		}
377	446		}
378	447

@@ -396,10 +465,12 @@ func (t *stdioTransport) request(ctx context.Context, id int64, method string, p
396	465		delete(t.pending, key)
397	466		t.mu.Unlock()
398	467		}
399		-	if err := t.write(ctx, frame); err != nil {
	468	+	if sent, err := t.writeFrame(ctx, frame); err != nil {
400	469		cleanup()
401	470		if ctx.Err() != nil {
402		-	t.cancelRequest(id, method, ctx.Err().Error())
	471	+	if sent {
	472	+	t.cancelRequest(id, method, ctx.Err().Error())
	473	+	}
403	474		return nil, ctx.Err()
404	475		}
405	476		return nil, err

@@ -494,7 +565,8 @@ func (t *stdioTransport) waitExit(d time.Duration) bool {
494	565
495	566		// close shuts the server down: close stdin (the MCP stdio shutdown signal)
496	567		// without waiting behind any blocked write, wait briefly, then terminate the
497		-	// process group. Descendants that outlive the direct child are signalled too.
	568	+	// process group. Group cleanup is one-shot and also runs when the direct
	569	+	// child exits on its own, so repeated or late calls never signal a stale id.
498	570		func (t *stdioTransport) close() error {
499	571		t.fail(errors.New("connection closed"))
500	572		_ = t.stdin.Close() // also unblocks a writer stuck on a full pipe

@@ -505,7 +577,7 @@ func (t *stdioTransport) close() error {
505	577		t.waitExit(time.Second)
506	578		}
507	579		}
508		-	t.reapGroup()
	580	+	t.guard.cleanup()
509	581		// With the child gone its stderr hits EOF once buffered data is read; give
510	582		// the reader a moment so diagnostics are complete. A descendant holding the
511	583		// pipe open only costs the timeout.

@@ -519,23 +591,5 @@ func (t *stdioTransport) close() error {
519	591		}
520	592		_ = t.outR.Close()
521	593		_ = t.errR.Close()
522		-	t.guard.release()
523	594		return nil
524	595		}
525		-
526		-	// reapGroup stops processes the direct child left behind. The process group
527		-	// outlives its leader, so this is meaningful even after the leader has been
528		-	// reaped (the group id cannot be reused while any member remains).
529		-	func (t *stdioTransport) reapGroup() {
530		-	if !t.guard.alive() {
531		-	return
532		-	}
533		-	t.guard.terminate(false)
534		-	deadline := time.Now().Add(300 * time.Millisecond)
535		-	for t.guard.alive() && time.Now().Before(deadline) {
536		-	time.Sleep(10 * time.Millisecond)
537		-	}
538		-	if t.guard.alive() {
539		-	t.guard.terminate(true)
540		-	}
541		-	}

 	"sync"
 	"sync/atomic"
 	"time"
+	"unicode/utf8"
 )
 // maxStdioFrame bounds one newline-delimited JSON-RPC message from a server.
 	// maxStderrLine truncates one retained stderr line; the rest of a longer
 	// line is still drained from the pipe, just not kept.
 	maxStderrLine = 4096
+	// maxStderrOverlap caps how far past maxStderrLine raw bytes are kept so
+	// a straddling secret can still be redacted.
+	maxStderrOverlap = 64 << 10
 	// writeQueueSize bounds frames waiting for the writer goroutine.
 	writeQueueSize = 64
 	// responseWriteTimeout bounds answering one server-initiated request.
 	ctx   context.Context
 	frame []byte
 	res   chan error
+	state atomic.Int32 // reqQueued, reqWriting, or reqSkipped
 }
+// A queued frame is claimed exactly once: by the writer (it will be written)
+// or by its caller (it will never be written).
+const (
+	reqQueued int32 = iota
+	reqWriting
+	reqSkipped
+)
 type stdioTransport struct {
 	name  string
 	cmd   *exec.Cmd
 	// cancelsSent counts notifications/cancelled sends started (diagnostics
 	// and tests).
 	cancelsSent atomic.Int64
+	// writesStarted counts frames the writer has begun writing (diagnostics
+	// and tests).
+	writesStarted atomic.Int64
 }
 // buildStdioEnv returns the restricted child environment.
 	}
 	outW.Close()
 	errW.Close()
+	guard, err := newProcessGuard(cmd)
+	if err != nil {
+		// Fail closed: never run a server that cannot be contained.
+		_ = stdin.Close()
+		_ = cmd.Process.Kill()
+		_ = cmd.Wait()
+		outR.Close()
+		errR.Close()
+		return nil, fmt.Errorf("start %s: %w", cfg.Command, red.scrubError(err))
+	}
 	ctx, cancel := context.WithCancel(context.Background())
 	t := &stdioTransport{
 		name:       name,
 		cmd:        cmd,
-		guard:      newProcessGuard(cmd),
+		guard:      guard,
 		red:        red,
 		ctx:        ctx,
 		cancel:     cancel,
 		case <-time.After(250 * time.Millisecond):
 		}
 		t.fail(fmt.Errorf("server process ended: %w", waitErr))
+		// The leader is gone: stop whatever it left behind, once, while the
+		// process-group id is still guaranteed to belong to our group.
+		t.guard.cleanup()
 	}()
 	return t, nil
 }
 // readStderr drains the child's stderr for its whole life, keeping a bounded
 // tail of bounded lines. It never stops early: a child blocked on a full
 // stderr pipe would hang.
+//
+// Redaction happens on the complete line before it is shortened for display.
+// For a line longer than the display limit the raw prefix is kept
+// maxSecretLen bytes past the cut, so a secret that straddles the cut is
+// matched whole; what remains of one anyway is trimmed afterwards.
 func (t *stdioTransport) readStderr(r io.Reader) {
 	defer close(t.stderrDone)
 	reader := bufio.NewReaderSize(r, 4096)
+	keep := maxStderrLine + min(t.red.maxLen, maxStderrOverlap)
 	var line []byte
 	truncated := false
 	flush := func() {
-		text := strings.TrimRight(string(line), "\r")
+		text := t.red.scrub(strings.TrimRight(string(line), "\r"))
-		if truncated {
+		if len(text) > maxStderrLine || truncated {
-			text += "…[line truncated]"
+			if len(text) > maxStderrLine {
+				cut := maxStderrLine
+				for cut > 0 && !utf8.RuneStart(text[cut]) {
+					cut--
+				}
+				text = text[:cut]
+			}
+			text = t.red.trimPartialSecret(text) + "…[line truncated]"
 		}
 		line, truncated = nil, false
 		t.stderrMu.Lock()
 	}
 	for {
 		chunk, isPrefix, err := reader.ReadLine()
-		if room := maxStderrLine - len(line); len(chunk) > room {
+		if room := keep - len(line); len(chunk) > room {
 			if room > 0 {
 				line = append(line, chunk[:room]...)
 			}
 		case <-t.doneCh:
 			return
 		case req := <-t.writeQ:
-			if err := req.ctx.Err(); err != nil {
+			if err := req.ctx.Err(); err != nil && req.state.CompareAndSwap(reqQueued, reqSkipped) {
 				req.res <- err // never started: nothing partial on the wire
 				continue
 			}
+			if !req.state.CompareAndSwap(reqQueued, reqWriting) {
+				continue // the caller gave up before we started
+			}
+			t.writesStarted.Add(1)
 			_, err := t.stdin.Write(req.frame)
 			req.res <- err
 			if err != nil {
 // write queues one frame (newline appended) and waits until it has been
 // written, ctx ends, or the connection fails.
 func (t *stdioTransport) write(ctx context.Context, frame []byte) error {
+	_, err := t.writeFrame(ctx, frame)
+	return err
+}
+// writeFrame is write that also reports whether any byte of the frame may
+// have reached the server. When ctx ends:
+//   - a frame the writer has not started is withdrawn and never written, and
+//     the transport stays usable;
+//   - a frame whose write is already in progress cannot be withdrawn (part of
+//     it may be on the wire and the rest could still be flushed later), so the
+//     transport is failed and stdin closed. The manager reconnects on the next
+//     call and the abandoned frame can never complete.
+func (t *stdioTransport) writeFrame(ctx context.Context, frame []byte) (sent bool, err error) {
 	req := &writeReq{ctx: ctx, frame: append(append(make([]byte, 0, len(frame)+1), frame...), '\n'), res: make(chan error, 1)}
 	select {
 	case t.writeQ <- req:
 	case <-ctx.Done():
-		return ctx.Err()
+		return false, ctx.Err()
 	case <-t.doneCh:
-		return t.failure()
+		return false, t.failure()
 	}
 	select {
 	case err := <-req.res:
-		return err
+		return req.state.Load() == reqWriting, err
 	case <-ctx.Done():
-		return ctx.Err()
+		if req.state.CompareAndSwap(reqQueued, reqSkipped) {
+			return false, ctx.Err()
+		}
+		select {
+		case err := <-req.res:
+			return req.state.Load() == reqWriting, err
+		default:
+		}
+		t.fail(fmt.Errorf("write to server interrupted: %w", ctx.Err()))
+		_ = t.stdin.Close() // unblocks the stuck write
+		return true, ctx.Err()
 	case <-t.doneCh:
 		select {
 		case err := <-req.res:
-			return err
+			return req.state.Load() == reqWriting, err
 		default:
 		}
-		return t.failure()
+		return req.state.Load() == reqWriting, t.failure()
 	}
 }
 		delete(t.pending, key)
 		t.mu.Unlock()
 	}
-	if err := t.write(ctx, frame); err != nil {
+	if sent, err := t.writeFrame(ctx, frame); err != nil {
 		cleanup()
 		if ctx.Err() != nil {
-			t.cancelRequest(id, method, ctx.Err().Error())
+			if sent {
+				t.cancelRequest(id, method, ctx.Err().Error())
+			}
 			return nil, ctx.Err()
 		}
 		return nil, err
 // close shuts the server down: close stdin (the MCP stdio shutdown signal)
 // without waiting behind any blocked write, wait briefly, then terminate the
-// process group. Descendants that outlive the direct child are signalled too.
+// process group. Group cleanup is one-shot and also runs when the direct
+// child exits on its own, so repeated or late calls never signal a stale id.
 func (t *stdioTransport) close() error {
 	t.fail(errors.New("connection closed"))
 	_ = t.stdin.Close() // also unblocks a writer stuck on a full pipe
 			t.waitExit(time.Second)
 		}
 	}
-	t.reapGroup()
+	t.guard.cleanup()
 	// With the child gone its stderr hits EOF once buffered data is read; give
 	// the reader a moment so diagnostics are complete. A descendant holding the
 	// pipe open only costs the timeout.
 	}
 	_ = t.outR.Close()
 	_ = t.errR.Close()
-	t.guard.release()
 	return nil
 }
-// reapGroup stops processes the direct child left behind. The process group
-// outlives its leader, so this is meaningful even after the leader has been
-// reaped (the group id cannot be reused while any member remains).
-func (t *stdioTransport) reapGroup() {
-	if !t.guard.alive() {
-		return
-	}
-	t.guard.terminate(false)
-	deadline := time.Now().Add(300 * time.Millisecond)
-	for t.guard.alive() && time.Now().Before(deadline) {
-		time.Sleep(10 * time.Millisecond)
-	}
-	if t.guard.alive() {
-		t.guard.terminate(true)
-	}
-}

M packages/agent/mcp/tools.go +34 -12


@@ -7,6 +7,7 @@ import (
7	7		"errors"
8	8		"fmt"
9	9		"strings"
	10	+	"unicode/utf8"
10	11
11	12		"github.com/bnema/zut/packages/core"
12	13		"github.com/bnema/zut/packages/provider"

@@ -283,35 +284,35 @@ func (t *listResourcesTool) Execute(ctx context.Context, args json.RawMessage, _
283	284		}
284	285		client, cfg, err := t.manager.client(ctx, server)
285	286		if err != nil {
286		-	out.printf("## %s\nunavailable: %s\n\n", sanitizeText(server, false), sanitizeText(firstLine(err.Error()), false))
	287	+	out.printf("## %s\nunavailable: %s\n\n", out.field(server), out.field(firstLine(err.Error())))
287	288		continue
288	289		}
289	290		if !client.Info().HasResources {
290	291		if in.Server != "" {
291		-	out.printf("## %s\nthis server does not offer resources\n\n", sanitizeText(server, false))
	292	+	out.printf("## %s\nthis server does not offer resources\n\n", out.field(server))
292	293		}
293	294		continue
294	295		}
295	296		reqCtx, cancel := withTimeout(ctx, cfg.ToolTimeout())
296	297		resources, rErr := client.ListResources(reqCtx)
297		-	out.printf("## %s\n", sanitizeText(server, false))
	298	+	out.printf("## %s\n", out.field(server))
298	299		if rErr != nil {
299		-	out.printf("resources error: %s\n", sanitizeText(rErr.Error(), false))
	300	+	out.printf("resources error: %s\n", out.field(rErr.Error()))
300	301		}
301	302		for _, r := range resources {
302	303		if out.full() {
303	304		break
304	305		}
305	306		var line strings.Builder
306		-	fmt.Fprintf(&line, "- %s", sanitizeText(r.URI, false))
	307	+	fmt.Fprintf(&line, "- %s", out.field(r.URI))
307	308		if name := firstNonEmpty(r.Title, r.Name); name != "" {
308		-	fmt.Fprintf(&line, " — %s", sanitizeText(name, false))
	309	+	fmt.Fprintf(&line, " — %s", out.field(name))
309	310		}
310	311		if r.MimeType != "" {
311		-	fmt.Fprintf(&line, " (%s)", sanitizeText(r.MimeType, false))
	312	+	fmt.Fprintf(&line, " (%s)", out.field(r.MimeType))
312	313		}
313	314		if r.Description != "" {
314		-	fmt.Fprintf(&line, ": %s", sanitizeText(r.Description, false))
	315	+	fmt.Fprintf(&line, ": %s", out.field(r.Description))
315	316		}
316	317		line.WriteByte('\n')
317	318		out.write(line.String())

@@ -324,19 +325,19 @@ func (t *listResourcesTool) Execute(ctx context.Context, args json.RawMessage, _
324	325		}
325	326		cancel()
326	327		if tErr != nil && !isMethodNotFound(tErr) {
327		-	out.printf("templates error: %s\n", sanitizeText(tErr.Error(), false))
	328	+	out.printf("templates error: %s\n", out.field(tErr.Error()))
328	329		}
329	330		for _, tpl := range templates {
330	331		if out.full() {
331	332		break
332	333		}
333	334		var line strings.Builder
334		-	fmt.Fprintf(&line, "- template %s", sanitizeText(tpl.URITemplate, false))
	335	+	fmt.Fprintf(&line, "- template %s", out.field(tpl.URITemplate))
335	336		if name := firstNonEmpty(tpl.Title, tpl.Name); name != "" {
336		-	fmt.Fprintf(&line, " — %s", sanitizeText(name, false))
	337	+	fmt.Fprintf(&line, " — %s", out.field(name))
337	338		}
338	339		if tpl.Description != "" {
339		-	fmt.Fprintf(&line, ": %s", sanitizeText(tpl.Description, false))
	340	+	fmt.Fprintf(&line, ": %s", out.field(tpl.Description))
340	341		}
341	342		line.WriteByte('\n')
342	343		out.write(line.String())

@@ -362,6 +363,27 @@ type budgetedText struct {
362	363
363	364		func (t *budgetedText) full() bool { return t.budget.truncated }
364	365
	366	+	// remaining is how many more bytes of text fit.
	367	+	func (t *budgetedText) remaining() int { return max(0, maxResultTextBytes-t.budget.used) }
	368	+
	369	+	// field prepares one server-supplied string for output: it is cut to the
	370	+	// remaining budget first (so a multi-megabyte description is never fully
	371	+	// processed) and then stripped of control sequences and line breaks.
	372	+	func (t *budgetedText) field(s string) string {
	373	+	return sanitizeText(clipUTF8(s, t.remaining()+1), false)
	374	+	}
	375	+
	376	+	// clipUTF8 shortens s to at most n bytes without splitting a rune.
	377	+	func clipUTF8(s string, n int) string {
	378	+	if len(s) <= n {
	379	+	return s
	380	+	}
	381	+	for n > 0 && !utf8.RuneStart(s[n]) {
	382	+	n--
	383	+	}
	384	+	return s[:n]
	385	+	}
	386	+
365	387		func (t *budgetedText) write(s string) {
366	388		if t.budget.truncated {
367	389		return

 	"errors"
 	"fmt"
 	"strings"
+	"unicode/utf8"
 	"github.com/bnema/zut/packages/core"
 	"github.com/bnema/zut/packages/provider"
 		}
 		client, cfg, err := t.manager.client(ctx, server)
 		if err != nil {
-			out.printf("## %s\nunavailable: %s\n\n", sanitizeText(server, false), sanitizeText(firstLine(err.Error()), false))
+			out.printf("## %s\nunavailable: %s\n\n", out.field(server), out.field(firstLine(err.Error())))
 			continue
 		}
 		if !client.Info().HasResources {
 			if in.Server != "" {
-				out.printf("## %s\nthis server does not offer resources\n\n", sanitizeText(server, false))
+				out.printf("## %s\nthis server does not offer resources\n\n", out.field(server))
 			}
 			continue
 		}
 		reqCtx, cancel := withTimeout(ctx, cfg.ToolTimeout())
 		resources, rErr := client.ListResources(reqCtx)
-		out.printf("## %s\n", sanitizeText(server, false))
+		out.printf("## %s\n", out.field(server))
 		if rErr != nil {
-			out.printf("resources error: %s\n", sanitizeText(rErr.Error(), false))
+			out.printf("resources error: %s\n", out.field(rErr.Error()))
 		}
 		for _, r := range resources {
 			if out.full() {
 				break
 			}
 			var line strings.Builder
-			fmt.Fprintf(&line, "- %s", sanitizeText(r.URI, false))
+			fmt.Fprintf(&line, "- %s", out.field(r.URI))
 			if name := firstNonEmpty(r.Title, r.Name); name != "" {
-				fmt.Fprintf(&line, " — %s", sanitizeText(name, false))
+				fmt.Fprintf(&line, " — %s", out.field(name))
 			}
 			if r.MimeType != "" {
-				fmt.Fprintf(&line, " (%s)", sanitizeText(r.MimeType, false))
+				fmt.Fprintf(&line, " (%s)", out.field(r.MimeType))
 			}
 			if r.Description != "" {
-				fmt.Fprintf(&line, ": %s", sanitizeText(r.Description, false))
+				fmt.Fprintf(&line, ": %s", out.field(r.Description))
 			}
 			line.WriteByte('\n')
 			out.write(line.String())
 		}
 		cancel()
 		if tErr != nil && !isMethodNotFound(tErr) {
-			out.printf("templates error: %s\n", sanitizeText(tErr.Error(), false))
+			out.printf("templates error: %s\n", out.field(tErr.Error()))
 		}
 		for _, tpl := range templates {
 			if out.full() {
 				break
 			}
 			var line strings.Builder
-			fmt.Fprintf(&line, "- template %s", sanitizeText(tpl.URITemplate, false))
+			fmt.Fprintf(&line, "- template %s", out.field(tpl.URITemplate))
 			if name := firstNonEmpty(tpl.Title, tpl.Name); name != "" {
-				fmt.Fprintf(&line, " — %s", sanitizeText(name, false))
+				fmt.Fprintf(&line, " — %s", out.field(name))
 			}
 			if tpl.Description != "" {
-				fmt.Fprintf(&line, ": %s", sanitizeText(tpl.Description, false))
+				fmt.Fprintf(&line, ": %s", out.field(tpl.Description))
 			}
 			line.WriteByte('\n')
 			out.write(line.String())
 func (t *budgetedText) full() bool { return t.budget.truncated }
+// remaining is how many more bytes of text fit.
+func (t *budgetedText) remaining() int { return max(0, maxResultTextBytes-t.budget.used) }
+// field prepares one server-supplied string for output: it is cut to the
+// remaining budget first (so a multi-megabyte description is never fully
+// processed) and then stripped of control sequences and line breaks.
+func (t *budgetedText) field(s string) string {
+	return sanitizeText(clipUTF8(s, t.remaining()+1), false)
+}
+// clipUTF8 shortens s to at most n bytes without splitting a rune.
+func clipUTF8(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	for n > 0 && !utf8.RuneStart(s[n]) {
+		n--
+	}
+	return s[:n]
+}
 func (t *budgetedText) write(s string) {
 	if t.budget.truncated {
 		return