fix: 透传 Claude 订阅审批上下文与服务端结果 (#794)

Co-authored-by: yizhuolin <yizhuolin@i-calorie.com>

internal/gateway/anthropic.go

@@ -79,8 +79,9 @@ type aRequest struct {
 			Schema json.RawMessage `json:"schema"`
 		} `json:"format,omitempty"`
 	} `json:"output_config,omitempty"`
-	Metadata json.RawMessage `json:"metadata,omitempty"`
-	Speed    string          `json:"speed,omitempty"` // "fast": Claude's fast mode
+	Metadata   json.RawMessage `json:"metadata,omitempty"`
+	Speed      string          `json:"speed,omitempty"` // "fast": Claude's fast mode
+	Safeguards json.RawMessage `json:"safeguards,omitempty"`
 }
 
 func parseAnthropic(body []byte) (*Request, error) {
@@ -90,6 +91,10 @@ func parseAnthropic(body []byte) (*Request, error) {
 	}
 	r := &Request{Model: a.Model, System: stringOrText(a.System), MaxTokens: a.MaxTokens,
 		Temp: a.Temperature, TopP: a.TopP, Stop: a.StopSequences, Stream: a.Stream, Fast: a.Speed == "fast"}
+	r.Safeguards = a.Safeguards
+	if string(r.Safeguards) == "null" {
+		r.Safeguards = nil
+	}
 	if len(a.Metadata) > 0 && string(a.Metadata) != "null" {
 		r.Metadata = a.Metadata
 	}
@@ -775,8 +780,12 @@ func (e *anthropicEncoder) finish() {
 	}
 	e.close()
 	res := e.col.finish()
+	delta := map[string]any{"stop_reason": stopToAnthropic(res.Stop), "stop_sequence": nil}
+	if len(res.SafeguardResults) > 0 {
+		delta["safeguard_results"] = res.SafeguardResults
+	}
 	e.w.event("message_delta", map[string]any{"type": "message_delta",
-		"delta": map[string]any{"stop_reason": stopToAnthropic(res.Stop), "stop_sequence": nil},
+		"delta": delta,
 		"usage": res.Usage.anthropic()})
 	e.w.event("message_stop", map[string]any{"type": "message_stop"})
 }
@@ -807,8 +816,12 @@ func renderAnthropic(res Result, model string) []byte {
 	if res.Model != "" {
 		model = res.Model
 	}
-	b, _ := json.Marshal(map[string]any{"id": id, "type": "message", "role": "assistant", "model": model,
-		"content": content, "stop_reason": stopToAnthropic(res.Stop), "stop_sequence": nil, "usage": res.Usage.anthropic()})
+	reply := map[string]any{"id": id, "type": "message", "role": "assistant", "model": model,
+		"content": content, "stop_reason": stopToAnthropic(res.Stop), "stop_sequence": nil, "usage": res.Usage.anthropic()}
+	if len(res.SafeguardResults) > 0 {
+		reply["safeguard_results"] = res.SafeguardResults
+	}
+	b, _ := json.Marshal(reply)
 	return b
 }
 

internal/gateway/automode_test.go

@@ -1,16 +1,23 @@
 package gateway
 
 import (
+	"bufio"
+	"context"
 	"encoding/json"
+	"fmt"
+	"io"
 	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"runtime"
+	"slices"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/yetone/magpie/internal/catalog"
 	"github.com/yetone/magpie/internal/provider"
+	"github.com/yetone/magpie/internal/wslrun"
 )
 
 // autoModeAsk is Claude Code's auto mode classifier's first stage as it
@@ -253,3 +260,269 @@ done
 		}
 	}
 }
+
+// The subscription bridge keeps the API's verdict opaque, including a denial,
+// skipped/unavailable calls and future fields. Missing results stay missing.
+func TestClaudeBridgeSafeguardResults(t *testing.T) {
+	for _, verdict := range []string{
+		`[{"type":"dangerous_tool_use","status":{"type":"available","tool_uses":{"toolu_client":{"type":"evaluated","outcome":"flagged","explanation":"dangerous"}}},"future":{"x":1}}]`,
+		`[{"type":"dangerous_tool_use","status":{"type":"available","tool_uses":{"toolu_client":{"type":"evaluated","outcome":"not_flagged"}}}}]`,
+		`[{"type":"dangerous_tool_use","status":{"type":"available","tool_uses":{"toolu_client":{"type":"skipped"}}}}]`,
+		`[{"type":"dangerous_tool_use","status":{"type":"unavailable","reason":"error"}}]`,
+		`[{"type":"future_check","status":{"type":"unknown"}}]`,
+		`null`,
+	} {
+		for _, whole := range []bool{false, true} {
+			name := "stream "
+			if whole {
+				name = "whole "
+			}
+			t.Run(name+verdict, func(t *testing.T) {
+				ch := make(chan Event, 16)
+				run := &subscriptionRun{segment: ch}
+				delta := map[string]any{"stop_reason": "tool_use"}
+				if verdict != "null" {
+					delta["safeguard_results"] = json.RawMessage(verdict)
+				}
+				d, _ := json.Marshal(map[string]any{"type": "stream_event", "event": map[string]any{"type": "message_delta", "delta": delta}})
+				lines := []string{
+					`{"type":"stream_event","event":{"type":"message_start","message":{"id":"msg_client","model":"claude-sonnet-5"}}}`,
+					`{"type":"stream_event","event":{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_client","name":"mcp__magpie__Bash"}}}`,
+					string(d),
+					`{"type":"stream_event","event":{"type":"message_stop"}}`,
+				}
+				if whole {
+					message := map[string]any{"id": "msg_whole", "model": "claude-sonnet-5", "stop_reason": "tool_use",
+						"content": []any{map[string]any{"type": "tool_use", "id": "toolu_client", "name": "mcp__magpie__Bash", "input": map[string]any{}}}}
+					if verdict != "null" {
+						message["safeguard_results"] = json.RawMessage(verdict)
+					}
+					d, _ := json.Marshal(map[string]any{"type": "assistant", "message": message})
+					lines = append(append([]string{}, cutAttempt...), string(d))
+				}
+				run.readOutput(strings.NewReader(strings.Join(lines, "\n")))
+				rec := httptest.NewRecorder()
+				enc := encoder(provider.Anthropic, newSSEWriter(rec), &Request{Model: "claude-sonnet-5"})
+				var col collector
+				for e := range ch {
+					enc.event(e)
+					col.add(e)
+				}
+				enc.finish()
+				result := col.finish()
+				var response map[string]any
+				json.Unmarshal(renderAnthropic(result, "claude-sonnet-5"), &response)
+				for _, e := range events(rec.Body.String()) {
+					if e["type"] == "message_delta" {
+						delta := e["delta"].(map[string]any)
+						actual, _ := json.Marshal(delta["safeguard_results"])
+						var wantValue any
+						json.Unmarshal([]byte(verdict), &wantValue)
+						want, _ := json.Marshal(wantValue)
+						if string(actual) != string(want) {
+							t.Fatalf("stream verdict changed: %s != %s", actual, want)
+						}
+					}
+				}
+				actual, _ := json.Marshal(response["safeguard_results"])
+				var wantValue any
+				json.Unmarshal([]byte(verdict), &wantValue)
+				want, _ := json.Marshal(wantValue)
+				if string(actual) != string(want) {
+					t.Fatalf("non-stream verdict changed: %s != %s", actual, want)
+				}
+				if !strings.Contains(rec.Body.String(), `"id":"toolu_client"`) || !strings.Contains(string(renderAnthropic(result, "claude-sonnet-5")), `"id":"toolu_client"`) {
+					t.Fatal("tool ID changed")
+				}
+			})
+		}
+	}
+}
+
+// Exercise the actual subprocess/control wire: changed and removed contexts
+// reach an existing run; a rejected update restarts it with the new context.
+func TestClaudeBridgeSafeguardContextUpdates(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("a Python script stands in for Claude Code")
+	}
+	t.Setenv("TMPDIR", t.TempDir())
+	dir := t.TempDir()
+	script := `#!/usr/bin/env python3
+import json,os,sys
+body=json.loads(os.environ.get('CLAUDE_CODE_EXTRA_BODY','{}'))
+effort=sys.argv[sys.argv.index('--effort')+1] if '--effort' in sys.argv else ''
+effort_id=''
+turns=0
+def send(x): print(json.dumps(x),flush=True)
+for line in sys.stdin:
+ x=json.loads(line)
+ if x['type']=='control_request':
+  if x['request']['subtype']=='initialize':
+   send({'type':'control_response','response':{'request_id':x['request_id'],'subtype':'success'}})
+   continue
+  settings=x['request']['settings']
+  if 'effortLevel' in settings:
+   effort=settings['effortLevel']
+   effort_id=x['request_id']
+   continue
+  if effort_id:
+   send({'type':'control_response','response':{'request_id':effort_id,'subtype':'success'}})
+  fresh=json.loads(x['request']['settings']['env']['CLAUDE_CODE_EXTRA_BODY'])
+  rejected=turns>0 and 'reject' in json.dumps(fresh)
+  if not rejected:body=fresh
+  send({'type':'control_response','response':{'request_id':'unrelated','subtype':'success'}})
+  send({'type':'control_response','response':{'request_id':x['request_id'],'subtype':'error' if rejected else 'success','error':'probe rejected'}})
+  continue
+ turns+=1
+ send({'type':'stream_event','event':{'type':'message_start','message':{'id':'m','model':'claude-sonnet-5'}}})
+ text=str(os.getpid())+' '+effort+' '+json.dumps(body,sort_keys=True)
+ send({'type':'stream_event','event':{'type':'content_block_delta','index':0,'delta':{'type':'text_delta','text':text}}})
+ send({'type':'stream_event','event':{'type':'message_delta','delta':{'stop_reason':'end_turn'}}})
+ send({'type':'stream_event','event':{'type':'message_stop'}})
+`
+	if err := os.WriteFile(filepath.Join(dir, "claude"), []byte(script), 0700); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	t.Setenv("CLAUDE_CODE_EXTRA_BODY", `{"model":"must-not-leak"}`)
+	s := New()
+	t.Cleanup(s.subscription.abortAll)
+	p := provider.Provider{ID: "claude", Account: &provider.Account{Agent: "claude", User: "u"}}
+	messages := []map[string]any{{"role": "assistant", "content": "seed"}, {"role": "user", "content": "hi"}}
+	var pid string
+	for i, context := range []string{"first", "second", "", "third", "reject"} {
+		b := map[string]any{"model": "claude-sonnet-5", "max_tokens": 64, "stream": false, "messages": messages}
+		effort := []string{"high", "low"}[i%2]
+		b["thinking"] = map[string]any{"type": "adaptive"}
+		b["output_config"] = map[string]any{"effort": effort}
+		if context != "" {
+			b["safeguards"] = []any{map[string]any{"type": "dangerous_tool_use", "classifier_context": map[string]any{"marker": context, "future": []int{1, 2}}}}
+		}
+		body, _ := json.Marshal(b)
+		req := httptest.NewRequest("POST", "/v1/messages", strings.NewReader(string(body)))
+		req.Header.Set("anthropic-beta", "claude-code-20250219,dangerous-tool-use-2026-09-03")
+		rec := httptest.NewRecorder()
+		var usage Usage
+		if code, msg := s.serveClaudeSubscription(rec, req, provider.Anthropic, p, "claude-sonnet-5", body, &usage); code != 200 {
+			t.Fatalf("%d %s", code, msg)
+		}
+		var a struct {
+			Content []struct {
+				Text string `json:"text"`
+			} `json:"content"`
+		}
+		if err := json.Unmarshal(rec.Body.Bytes(), &a); err != nil || len(a.Content) != 1 {
+			t.Fatalf("bad reply %s", rec.Body)
+		}
+		text := a.Content[0].Text
+		parts := strings.SplitN(text, " ", 3)
+		if len(parts) != 3 {
+			t.Fatalf("no pid/context: %s", text)
+		}
+		if parts[1] != effort {
+			t.Fatalf("stale effort: %s, want %s", parts[1], effort)
+		}
+		if pid != "" && (parts[0] == pid) == (context == "reject") {
+			t.Fatalf("wrong reuse for %q: %s => %s", context, pid, parts[0])
+		}
+		pid = parts[0]
+		var extra map[string]json.RawMessage
+		if err := json.Unmarshal([]byte(parts[2]), &extra); err != nil {
+			t.Fatal(err)
+		}
+		expected := safeguardBody(nil)
+		if v := b["safeguards"]; v != nil {
+			raw, _ := json.Marshal(v)
+			expected = safeguardBody(raw)
+		}
+		got, _ := json.Marshal(extra)
+		if string(got) != expected {
+			t.Fatalf("stale/modified context: %s != %s", got, expected)
+		}
+		messages = append(messages, map[string]any{"role": "assistant", "content": text}, map[string]any{"role": "user", "content": "next"})
+	}
+}
+
+// The beta must reach a Claude Code running in WSL as well as a native run.
+func TestClaudeBridgeSafeguardWSLEnv(t *testing.T) {
+	cli := claudeCLI{wsl: &wslrun.Tool{}}
+	env := cli.env([]string{"ANTHROPIC_BETAS=dangerous-tool-use-2026-09-03"}, "")
+	if !slices.Contains(env, "ANTHROPIC_BETAS=dangerous-tool-use-2026-09-03") ||
+		!slices.Contains(env, "WSLENV=ANTHROPIC_BETAS") {
+		t.Fatalf("safety beta not forwarded to WSL: %v", env)
+	}
+}
+
+// A context update and its tool-result handoff own the run together, through
+// the reply's end. Another caller cannot update it or abort its owner.
+func TestClaudeBridgeSafeguardConcurrentHandoff(t *testing.T) {
+	s := New()
+	b := s.subscription
+	input, stdin := io.Pipe()
+	stdout, output := io.Pipe()
+	run := &subscriptionRun{bridge: b, token: "tok", stdin: stdin, safeguardBeta: "dangerous-tool-use-2026-09-03",
+		pending: map[string]chan mcpToolResult{"a": make(chan mcpToolResult, 1), "b": make(chan mcpToolResult, 1)}}
+	b.runs[run.token], b.calls["a"], b.calls["b"] = run, run, run
+	t.Cleanup(func() { b.abortAll(); stdin.Close(); input.Close(); output.Close(); stdout.Close() })
+	go run.readOutput(stdout)
+	go func() {
+		scanner := bufio.NewScanner(input)
+		for scanner.Scan() {
+			var ctl struct {
+				RequestID string `json:"request_id"`
+			}
+			if json.Unmarshal(scanner.Bytes(), &ctl) == nil {
+				fmt.Fprintf(output, "{\"type\":\"control_response\",\"response\":{\"subtype\":\"success\",\"request_id\":%q}}\n", ctl.RequestID)
+			}
+		}
+	}()
+	req := &Request{SafeguardBeta: run.safeguardBeta, Safeguards: json.RawMessage(`[{"context":"A"}]`),
+		Messages: []Message{{Role: "assistant"}, {Role: "user", Parts: []Part{{Kind: ToolResult, CallID: "a", Text: "A"}}}}}
+	found, results := b.findRun(req)
+	if found != run || !found.claimResume() {
+		t.Fatal("first claim failed")
+	}
+	if err := run.setSafeguards(req); err != nil {
+		t.Fatal(err)
+	}
+	run.endSegment() // a late end with no attached segment cannot release A
+	// B arrives after A's ACK and before its handoff, then again while A's
+	// reply is active, using another call that still maps to the same run.
+	blocked := func(id string) {
+		t.Helper()
+		body := []byte(`{"model":"claude-sonnet-5","max_tokens":64,"safeguards":[{"context":"B"}],"messages":[{"role":"assistant","content":"call"},{"role":"user","content":[{"type":"tool_result","tool_use_id":"` + id + `","content":"B"}]}]}`)
+		done := make(chan int, 1)
+		go func() {
+			rec := httptest.NewRecorder()
+			code, _ := s.serveSubscription(rec, httptest.NewRequest("POST", "/v1/messages", nil), provider.Anthropic, "Claude Code", req.Model, body, &Usage{},
+				func(context.Context, *Request) (*subscriptionRun, <-chan Event, error) {
+					return nil, nil, fmt.Errorf("must not start another run")
+				})
+			done <- code
+		}()
+		select {
+		case code := <-done:
+			if code != 409 || run.closed || string(run.safeguards) != string(req.Safeguards) {
+				t.Fatalf("second caller: status %d, closed %v, context %s", code, run.closed, run.safeguards)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatal("second caller did not reject the claimed run")
+		}
+	}
+	blocked("a")
+	waiter := run.pending["a"]
+	segment, err := run.continueWith(results, nil)
+	if err != nil || (<-waiter).Content[0]["text"] != "A" {
+		t.Fatalf("first handoff: %v", err)
+	}
+	blocked("b")
+	fmt.Fprintln(output, `{"type":"stream_event","event":{"type":"message_start","message":{"id":"m"}}}`)
+	fmt.Fprintln(output, `{"type":"stream_event","event":{"type":"message_delta","delta":{"stop_reason":"tool_use"}}}`)
+	fmt.Fprintln(output, `{"type":"stream_event","event":{"type":"message_stop"}}`)
+	for range segment {
+	}
+	req.Messages[1].Parts[0].CallID = "b"
+	if next, _ := b.findRun(req); next != run || !next.claimResume() {
+		t.Fatal("claim not released at segment end")
+	}
+}

internal/gateway/claude_subscription.go

@@ -30,6 +30,7 @@ package gateway
 
 import (
 	"bufio"
+	"bytes"
 	"context"
 	"crypto/rand"
 	"crypto/sha256"
@@ -125,12 +126,19 @@ type subscriptionRun struct {
 	// the tools its agent was told of, as it started and since
 	tools map[string]bool
 
-	mu      sync.Mutex
-	segment chan Event
-	pending map[string]chan mcpToolResult
-	closed  bool
-	stderr  strings.Builder
-	timer   *time.Timer
+	mu       sync.Mutex
+	segment  chan Event
+	pending  map[string]chan mcpToolResult
+	closed   bool
+	resuming bool // a tool-result request owns this run until its segment ends
+	stderr   strings.Builder
+
+	// Only one settings update is in flight; it must be acknowledged before
+	// a tool result or a new user turn can trigger the next model request.
+	settingsMu    sync.Mutex
+	safeguards    json.RawMessage
+	safeguardBeta string
+	timer         *time.Timer
 
 	// Claude Code's input, kept open for the next turn; owner is the account
 	// it runs as, and idleKey and idleAt say it is waiting for one. convKey
@@ -365,7 +373,7 @@ func (c claudeCLI) command(ctx context.Context, args ...string) *exec.Cmd {
 // claudeEnvVars are the variables magpie sets for a Claude Code run, which
 // one in WSL is handed (wslrun's Env).
 var claudeEnvVars = []string{"ENABLE_CLAUDEAI_MCP_SERVERS", "DISABLE_AUTO_COMPACT", "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC",
-	"DISABLE_TELEMETRY", "DISABLE_ERROR_REPORTING", "DISABLE_AUTOUPDATER"}
+	"DISABLE_TELEMETRY", "DISABLE_ERROR_REPORTING", "DISABLE_AUTOUPDATER", "ANTHROPIC_BETAS"}
 
 // env is env for a run in config directory configDir ("" for the account
 // Claude Code is signed in to): for one in WSL, what of it goes in.
@@ -470,7 +478,11 @@ func (b *subscriptionBridge) start(ctx context.Context, req *Request, model, con
 	}
 	cmd := binary.command(context.Background(), args...)
 	cmd.Dir = work
-	cmd.Env = binary.env(netproxy.EnvWith(claudeProxy(ctx), cleanClaudeEnv(os.Environ())), configDir)
+	env := netproxy.EnvWith(claudeProxy(ctx), cleanClaudeEnv(os.Environ()))
+	if len(req.Safeguards) > 0 {
+		env = append(env, "ANTHROPIC_BETAS="+req.SafeguardBeta)
+	}
+	cmd.Env = binary.env(env, configDir)
 	stdin, err := cmd.StdinPipe()
 	if err != nil {
 		cleanup()
@@ -488,6 +500,7 @@ func (b *subscriptionBridge) start(ctx context.Context, req *Request, model, con
 	}
 
 	run := &subscriptionRun{bridge: b, token: token, model: model, cmd: cmd, tmp: tmp, schema: len(req.Schema) > 0, pending: map[string]chan mcpToolResult{}, stdin: stdin, owner: owner, effort: req.Effort, sessions: sessions}
+	run.safeguardBeta = req.SafeguardBeta
 	if user, _ := ownerAccount(owner); user != "" {
 		run.loginVersion = provider.ClaudeLoginVersion(user)
 	}
@@ -508,6 +521,18 @@ func (b *subscriptionBridge) start(ctx context.Context, req *Request, model, con
 	}()
 	go run.readOutput(stdout)
 
+	if len(req.Safeguards) > 0 {
+		// Initialize the SDK before applying settings. Send the potentially
+		// large context over stdin rather than an environment entry.
+		if err := run.cliControl(map[string]any{"subtype": "initialize"}); err != nil {
+			run.abort()
+			return nil, nil, err
+		}
+	}
+	if err := run.setSafeguards(req); err != nil {
+		run.abort()
+		return nil, nil, err
+	}
 	var prompt []map[string]any
 	if from != nil {
 		// the session is the run's now: its file goes when the run does,
@@ -529,6 +554,89 @@ func (b *subscriptionBridge) start(ctx context.Context, req *Request, model, con
 	return run, segment, nil
 }
 
+// safeguardBody passes only the safety field through EXTRA_BODY: overriding the
+// CLI's own betas there would remove headers its genuine request needs.
+func safeguardBody(raw json.RawMessage) string {
+	body := map[string]any{}
+	if len(raw) > 0 {
+		body["safeguards"] = raw
+	}
+	b, _ := json.Marshal(body)
+	return string(b)
+}
+
+// setSafeguards updates the caller's context, not the VPS's permission state.
+// A CLI unable to acknowledge the update is discarded by the caller; its old
+// context must never be used for another turn. The client's approval still
+// decides whether the tool runs, including a server verdict that is unavailable.
+func (r *subscriptionRun) setSafeguards(req *Request) error {
+	r.settingsMu.Lock()
+	defer r.settingsMu.Unlock()
+	if bytes.Equal(r.safeguards, req.Safeguards) && (len(req.Safeguards) == 0 || r.safeguardBeta == req.SafeguardBeta) {
+		return nil
+	}
+	if len(req.Safeguards) > 0 && r.safeguardBeta != req.SafeguardBeta {
+		return errors.New("safety beta changed; restart Claude Code")
+	}
+	err := r.cliControl(map[string]any{"subtype": "apply_flag_settings", "settings": map[string]any{
+		"env": map[string]string{"CLAUDE_CODE_EXTRA_BODY": safeguardBody(req.Safeguards)}}})
+	if err == nil {
+		r.safeguards = bytes.Clone(req.Safeguards)
+	}
+	return err
+}
+
+// cliControl shares the SDK reply dispatcher with rewind, so unrelated or
+// late control responses cannot acknowledge a safety-context update.
+func (r *subscriptionRun) cliControl(request map[string]any) error {
+	id := randomToken()
+	ch := make(chan controlReply, 1)
+	r.mu.Lock()
+	if r.closed {
+		r.mu.Unlock()
+		return errors.New("Claude Code ended")
+	}
+	if r.controls == nil {
+		r.controls = map[string]chan controlReply{}
+	}
+	r.controls[id] = ch
+	r.mu.Unlock()
+	defer func() {
+		r.mu.Lock()
+		delete(r.controls, id)
+		r.mu.Unlock()
+	}()
+	line, _ := json.Marshal(map[string]any{"type": "control_request", "request_id": id, "request": request})
+	if _, err := r.stdin.Write(append(line, '\n')); err != nil {
+		return err
+	}
+	t := time.NewTimer(10 * time.Second)
+	defer t.Stop()
+	select {
+	case reply, ok := <-ch:
+		if !ok {
+			return errors.New("Claude Code ended")
+		}
+		if reply.Subtype != "success" {
+			return fmt.Errorf("Claude settings: %s", reply.Error)
+		}
+		return nil
+	case <-t.C:
+		return errors.New("Claude Code did not acknowledge safety context")
+	}
+}
+
+// claimResume holds a tool-result handoff across its context update and reply.
+func (r *subscriptionRun) claimResume() bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.closed || r.resuming || r.segment != nil {
+		return false
+	}
+	r.resuming = true
+	return true
+}
+
 // resume hands a conversation's next turn to the run that had its last
 // one, when one is waiting: the user's messages since, and nothing else.
 func (b *subscriptionBridge) resume(req *Request, owner string) (*subscriptionRun, <-chan Event) {
@@ -554,6 +662,10 @@ func (b *subscriptionBridge) resume(req *Request, owner string) (*subscriptionRu
 	if rewinding != nil {
 		<-rewinding
 	}
+	if err := run.setSafeguards(req); err != nil {
+		run.abort()
+		return nil, nil
+	}
 	// the message's id is the turn's, to rewind to if the client gives up
 	// on it
 	turn := newUUID()
@@ -1226,6 +1338,7 @@ func cleanClaudeEnv(env []string) []string {
 		"ANTHROPIC_BASE_URL": true, "ANTHROPIC_API_KEY": true, "ANTHROPIC_AUTH_TOKEN": true,
 		"CLAUDECODE": true, "CLAUDE_CODE_ENTRYPOINT": true, "CLAUDE_CODE_SSE_PORT": true,
 		"CLAUDE_CODE_OAUTH_TOKEN": true, "CLAUDE_CODE_PROMPT_CACHE_TTL": true,
+		"CLAUDE_CODE_EXTRA_BODY": true, "ANTHROPIC_BETAS": true,
 	}
 	out := make([]string, 0, len(env)+3)
 	for _, e := range env {
@@ -1292,6 +1405,9 @@ func (r *subscriptionRun) endSegment() {
 	r.mu.Lock()
 	ch := r.segment
 	r.segment = nil
+	if ch != nil {
+		r.resuming = false
+	}
 	r.mu.Unlock()
 	if ch != nil {
 		close(ch)
@@ -1460,12 +1576,13 @@ func (r *subscriptionRun) readOutput(rd io.Reader) {
 					Text string `json:"text"`
 				} `json:"content_block"`
 				Delta struct {
-					Type        string `json:"type"`
-					Text        string `json:"text"`
-					Thinking    string `json:"thinking"`
-					Signature   string `json:"signature"`
-					PartialJSON string `json:"partial_json"`
-					StopReason  string `json:"stop_reason"`
+					Type             string          `json:"type"`
+					Text             string          `json:"text"`
+					Thinking         string          `json:"thinking"`
+					Signature        string          `json:"signature"`
+					PartialJSON      string          `json:"partial_json"`
+					StopReason       string          `json:"stop_reason"`
+					SafeguardResults json.RawMessage `json:"safeguard_results"`
 				} `json:"delta"`
 				Usage cliUsage `json:"usage"`
 			} `json:"event"`
@@ -1474,11 +1591,12 @@ func (r *subscriptionRun) readOutput(rd io.Reader) {
 			return
 		}
 		var whole struct {
-			ID         string   `json:"id"`
-			Model      string   `json:"model"`
-			StopReason string   `json:"stop_reason"`
-			Usage      cliUsage `json:"usage"`
-			Content    []struct {
+			ID               string          `json:"id"`
+			Model            string          `json:"model"`
+			StopReason       string          `json:"stop_reason"`
+			Usage            cliUsage        `json:"usage"`
+			SafeguardResults json.RawMessage `json:"safeguard_results"`
+			Content          []struct {
 				Type      string          `json:"type"`
 				Text      string          `json:"text"`
 				Thinking  string          `json:"thinking"`
@@ -1598,7 +1716,7 @@ func (r *subscriptionRun) readOutput(rd io.Reader) {
 					}
 				}
 			}
-			r.emit(Event{Kind: KUsage, Usage: usage(whole.Usage), RequestID: reqID})
+			r.emit(Event{Kind: KUsage, Usage: usage(whole.Usage), RequestID: reqID, SafeguardResults: whole.SafeguardResults})
 			if quiet != nil {
 				quiet.Stop()
 			}
@@ -1657,7 +1775,7 @@ func (r *subscriptionRun) readOutput(rd io.Reader) {
 				}
 			}
 		case "message_delta":
-			r.emit(Event{Kind: KUsage, Usage: usage(e.Usage), RequestID: reqID})
+			r.emit(Event{Kind: KUsage, Usage: usage(e.Usage), RequestID: reqID, SafeguardResults: e.Delta.SafeguardResults})
 			if e.Delta.StopReason != "" {
 				stopped = true
 			}
@@ -2299,6 +2417,7 @@ func (r *subscriptionRun) finish() {
 		return
 	}
 	r.closed = true
+	r.resuming = false
 	if r.timer != nil {
 		r.timer.Stop()
 	}
@@ -2445,6 +2564,15 @@ func (s *Server) serveSubscription(w http.ResponseWriter, r *http.Request, from
 		return writeError(w, from, 400, err.Error()), err.Error()
 	}
 	req.Model = model
+	if len(req.Safeguards) > 0 {
+		req.SafeguardBeta = "dangerous-tool-use-2026-09-03"
+		for _, beta := range strings.Split(r.Header.Get("anthropic-beta"), ",") {
+			if beta = strings.TrimSpace(beta); strings.HasPrefix(beta, "dangerous-tool-use-") {
+				req.SafeguardBeta = beta
+				break
+			}
+		}
+	}
 	// its model searches the web with magpie's tool, which magpie answers
 	var search Tool
 	if req.WebSearch && !searching(r.Context()) {
@@ -2455,6 +2583,10 @@ func (s *Server) serveSubscription(w http.ResponseWriter, r *http.Request, from
 	}
 
 	run, results, how := s.subscription.match(req)
+	if run != nil && !run.claimResume() {
+		msg := "the agent's turn is already being resumed"
+		return writeError(w, from, http.StatusConflict, msg), msg
+	}
 	// the client rewrote the conversation since the run's last reply, as Pi
 	// does compacting it mid-turn: the run's agent holds the one from before,
 	// would answer from it, and tell the client a context as large as ever,
@@ -2476,7 +2608,14 @@ func (s *Server) serveSubscription(w http.ResponseWriter, r *http.Request, from
 	}
 	var events <-chan Event
 	if run != nil {
-		if events, err = run.continueWith(results, more); err != nil {
+		if name == "Claude Code" {
+			err = run.setSafeguards(req)
+		}
+		if err == nil {
+			events, err = run.continueWith(results, more)
+		}
+		if err != nil {
+			run.abort()
 			// it ended while it waited: a new one is told the whole
 			// conversation
 			run, how = nil, runExpired

internal/gateway/ir.go

@@ -127,6 +127,9 @@ type Request struct {
 	// Anthropic upstream: a relay that serves only Claude Code turns a
 	// request without it away (#359).
 	Metadata json.RawMessage
+	// Safeguards are the caller's safety context, opaque to the gateway.
+	Safeguards    json.RawMessage
+	SafeguardBeta string
 	// Schema is the JSON schema an Anthropic client asked the answer to fit
 	// (output_config.format, of type json_schema).
 	Schema json.RawMessage
@@ -180,10 +183,11 @@ type Event struct {
 	// Code: for KError, the source error or safety-filter code (rate_limit,
 	// server_error, bio_policy, content_filter…); RequestID: the vendor's id for the
 	// request the event is of, when it is known by then
-	Code      string
-	RequestID string
-	Usage     Usage
-	Hits      []Hit
+	Code             string
+	RequestID        string
+	Usage            Usage
+	Hits             []Hit
+	SafeguardResults json.RawMessage
 }
 
 // Usage counts tokens.
@@ -244,11 +248,12 @@ func (u *Usage) add(v Usage) {
 
 // Result is a whole reply, for non-streaming clients.
 type Result struct {
-	ID    string
-	Model string
-	Parts []Part
-	Stop  string
-	Usage Usage
+	ID               string
+	Model            string
+	Parts            []Part
+	Stop             string
+	Usage            Usage
+	SafeguardResults json.RawMessage
 }
 
 // collector assembles a Result from events. Encoders use the same logic to
@@ -281,6 +286,9 @@ func (c *collector) closeTool() {
 }
 
 func (c *collector) add(ev Event) {
+	if len(ev.SafeguardResults) > 0 {
+		c.res.SafeguardResults = ev.SafeguardResults
+	}
 	switch ev.Kind {
 	case KStart:
 		c.res.ID, c.res.Model = ev.MsgID, ev.Model