From a3ddffaadeddc552df2d5baee57223a86dcc2ca0 Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Wed, 26 Aug 2026 18:16:01 +0000 Subject: [PATCH 1/6] feat(git-agent): delegate caller tools to remote agents Issue task-bound caller-tool capabilities from the supervisor and deliver them through the authenticated HTTPS sidecar control channel without placing credentials in Git protocol data. Restrict discovery and execution to the parent-authorized allowlist, recheck task and agent bindings on every call, and revoke or expire capabilities with secret-free audit events. Amp-Thread-ID: https://ampcode.com/threads/T-01a03e3c-39ce-714c-b7fe-60010c4f452f --- pkg/ai/callertools/runtime.go | 338 ++++++++++++++++++- pkg/ai/client.go | 12 +- pkg/ai/remote_provider.go | 106 ++++++ pkg/aichat/service.go | 7 + pkg/api/runtime_config.go | 27 ++ pkg/api/sandbox_ref.go | 26 +- pkg/api/sandbox_registry.go | 6 + pkg/cli/ai_prompt_file.go | 1 + pkg/cli/ai_sandbox.go | 2 + pkg/cli/ai_sandbox_remote.go | 5 +- pkg/cli/gitagent_hook.go | 1 + pkg/cli/gitagent_runtask.go | 24 +- pkg/cli/gitagent_serve.go | 2 +- pkg/cli/gitagent_serve_https.go | 66 ++++ pkg/cli/serve.go | 3 + pkg/cli/serve_auth.go | 18 +- pkg/gitagent/callertools.go | 581 ++++++++++++++++++++++++++++++++ pkg/gitagent/dispatch.go | 3 + pkg/gitagent/hookmain.go | 1 + pkg/gitagent/httpclient.go | 40 +++ pkg/sandbox/adapter/gitagent.go | 62 ++++ 21 files changed, 1308 insertions(+), 23 deletions(-) create mode 100644 pkg/ai/remote_provider.go create mode 100644 pkg/gitagent/callertools.go diff --git a/pkg/ai/callertools/runtime.go b/pkg/ai/callertools/runtime.go index f85643ae..228b3480 100644 --- a/pkg/ai/callertools/runtime.go +++ b/pkg/ai/callertools/runtime.go @@ -11,8 +11,10 @@ import ( "encoding/base64" "encoding/json" "fmt" + "io" "net" "net/http" + "net/url" "strings" "sync" "sync/atomic" @@ -20,6 +22,7 @@ import ( aitools "github.com/flanksource/captain/pkg/ai/tools" "github.com/flanksource/captain/pkg/api" + "github.com/flanksource/commons/logger" "github.com/mark3labs/mcp-go/mcp" "github.com/mark3labs/mcp-go/server" "github.com/santhosh-tekuri/jsonschema/v6" @@ -29,12 +32,32 @@ const ( endpointPath = "/mcp" serverName = "captain" defaultApprovalTimeout = 5 * time.Minute + maxMCPRequestBytes = 4 << 20 + // RemoteEndpointPrefix is mounted by captain serve for task-scoped caller + // tool capabilities. The random capability ID selects authority; the bearer + // secret still authenticates it. + RemoteEndpointPrefix = "/caller-tools/" + TaskHeader = "X-Captain-Task" + AgentHeader = "X-Captain-Agent" // ToolUseIDInputKey carries an out-of-process provider's tool call ID // through clients that cannot attach MCP request metadata. The runtime // removes it before policy, schema validation, and handler execution. ToolUseIDInputKey = "__captain_tool_use_id" ) +var callertoolsLog = logger.GetLogger("ai") + +// AuditEvent records a capability decision without request inputs, results, or +// bearer values. +type AuditEvent struct { + Action string + TaskID string + Agent string + Tool string + Result string + Reason string +} + // Options defines one private caller-tool capability. type Options struct { Definitions []api.ToolDefinition @@ -47,6 +70,7 @@ type Options struct { // ValidateCredential rechecks the persisted lease on every request and // immediately before a tool handler executes. ValidateCredential func(context.Context) error + Audit func(AuditEvent) ApprovalTimeout time.Duration } @@ -62,6 +86,7 @@ type Runtime struct { token string tokenHash [sha256.Size]byte expiresAt time.Time + audit func(AuditEvent) approvalTimeout time.Duration ctx context.Context @@ -71,11 +96,33 @@ type Runtime struct { endpoint api.CallerToolEndpoint server *http.Server listener net.Listener + mcpHTTP http.Handler closeOnce sync.Once closeErr error } +type delegationContextKey struct{} + +type delegatedCapability struct { + runtime *Runtime + id string + tokenHash [sha256.Size]byte + binding api.CallerToolBinding + tools map[string]struct{} + ctx context.Context + cancel context.CancelFunc + revoked atomic.Bool + expired atomic.Bool +} + +var remoteCapabilities = struct { + sync.RWMutex + values map[string]*delegatedCapability +}{values: map[string]*delegatedCapability{}} + +var remoteHandlerEnabled atomic.Bool + // New validates and resolves the tool policy before starting a private server. func New(options Options) (*Runtime, error) { if !options.ExpiresAt.IsZero() && !options.ExpiresAt.After(time.Now()) { @@ -108,6 +155,7 @@ func New(options Options) (*Runtime, error) { schemas: make(map[string]*jsonschema.Schema, len(definitions)), canUseTool: options.CanUseTool, validate: options.ValidateCredential, + audit: options.Audit, sessionID: options.SessionID, token: token, tokenHash: sha256.Sum256([]byte(token)), @@ -140,8 +188,9 @@ func New(options Options) (*Runtime, error) { server.WithStateLess(true), server.WithEndpointPath(endpointPath), ) + runtime.mcpHTTP = runtime.guardToolName(handler) runtime.server = &http.Server{ - Handler: runtime.authorize(handler), + Handler: runtime.authorizeLocal(runtime.mcpHTTP), ReadHeaderTimeout: 5 * time.Second, } runtime.endpoint = api.CallerToolEndpoint{ @@ -150,6 +199,7 @@ func New(options Options) (*Runtime, error) { Headers: map[string]string{ "Authorization": "Bearer " + token, }, + Delegate: runtime.delegate, } go func() { _ = runtime.server.Serve(listener) @@ -164,6 +214,28 @@ func (r *Runtime) Endpoint() api.CallerToolEndpoint { return endpoint } +// RemoteHandler serves delegated capabilities from the Captain supervisor. +// Mount it at RemoteEndpointPrefix outside generic API-token middleware: each +// request is authenticated by its own task capability instead. +func RemoteHandler() http.Handler { + remoteHandlerEnabled.Store(true) + return http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + id, ok := remoteCapabilityID(request.URL.Path) + if !ok { + http.NotFound(w, request) + return + } + remoteCapabilities.RLock() + capability := remoteCapabilities.values[id] + remoteCapabilities.RUnlock() + if capability == nil { + writeCredentialRejection(w) + return + } + capability.serveHTTP(w, request) + }) +} + // CredentialHash returns the SHA-256 bearer hash persisted by an authority. // The plaintext capability remains confined to Endpoint headers. func (r *Runtime) CredentialHash() []byte { @@ -187,13 +259,101 @@ func (r *Runtime) Close() error { func (r *Runtime) Revoke() { if r.revoked.CompareAndSwap(false, true) { r.cancel() + var capabilities []*delegatedCapability + remoteCapabilities.Lock() + for id, capability := range remoteCapabilities.values { + if capability.runtime == r { + delete(remoteCapabilities.values, id) + capabilities = append(capabilities, capability) + } + } + remoteCapabilities.Unlock() + for _, capability := range capabilities { + capability.revoke("parent_revoked") + } + } +} + +func (r *Runtime) delegate(_ context.Context, binding api.CallerToolBinding) (*api.CallerToolDelegation, error) { + if !remoteHandlerEnabled.Load() { + return nil, fmt.Errorf("remote caller tools require captain serve to mount the authenticated talkback handler") } + if strings.TrimSpace(binding.TaskID) == "" || strings.TrimSpace(binding.Agent) == "" { + return nil, fmt.Errorf("remote caller-tool capability requires task and agent bindings") + } + if len(binding.ToolNames) == 0 { + return nil, fmt.Errorf("remote caller-tool capability requires at least one delegated tool") + } + if !binding.ExpiresAt.After(time.Now()) { + return nil, fmt.Errorf("remote caller-tool capability expiry must be in the future") + } + if err := r.validateActive(context.Background()); err != nil { + return nil, err + } + tools := make(map[string]struct{}, len(binding.ToolNames)) + for _, name := range binding.ToolNames { + name = strings.TrimSpace(name) + if _, authorized := r.definitions[name]; authorized { + tools[name] = struct{}{} + } else { + r.auditEvent(binding, AuditEvent{ + Action: "issuance", Tool: name, Result: "denied", Reason: "parent_not_authorized", + }) + } + } + if len(tools) == 0 { + return nil, fmt.Errorf("none of the requested caller tools are authorized by the supervisor") + } + id, err := randomID(18) + if err != nil { + return nil, fmt.Errorf("generate remote caller-tool capability ID: %w", err) + } + secret, err := randomID(32) + if err != nil { + return nil, fmt.Errorf("generate remote caller-tool credential: %w", err) + } + token := "cap_" + id + "." + secret + capabilityCtx, capabilityCancel := context.WithCancel(r.ctx) + capability := &delegatedCapability{ + runtime: r, id: id, tokenHash: sha256.Sum256([]byte(token)), binding: binding, + tools: tools, ctx: capabilityCtx, cancel: capabilityCancel, + } + remoteCapabilities.Lock() + if !r.active() { + remoteCapabilities.Unlock() + capabilityCancel() + return nil, fmt.Errorf("caller-tool credential is inactive") + } + remoteCapabilities.values[id] = capability + remoteCapabilities.Unlock() + time.AfterFunc(time.Until(binding.ExpiresAt), capability.expire) + r.auditEvent(binding, AuditEvent{Action: "issuance", Result: "issued"}) + delegation := &api.CallerToolDelegation{ + Endpoint: api.CallerToolEndpoint{ + Name: serverName, + URL: "http://127.0.0.1" + RemoteEndpointPrefix + id + endpointPath, + Headers: map[string]string{ + "Authorization": "Bearer " + token, + TaskHeader: binding.TaskID, + AgentHeader: binding.Agent, + }, + }, + Revoke: func() { + remoteCapabilities.Lock() + if current := remoteCapabilities.values[id]; current == capability { + delete(remoteCapabilities.values, id) + } + remoteCapabilities.Unlock() + capability.revoke("task_completed") + }, + } + return delegation, nil } -func (r *Runtime) filterTools(_ context.Context, tools []mcp.Tool) []mcp.Tool { +func (r *Runtime) filterTools(ctx context.Context, tools []mcp.Tool) []mcp.Tool { filtered := make([]mcp.Tool, 0, len(tools)) for _, tool := range tools { - if _, ok := r.definitions[tool.Name]; ok { + if r.toolAuthorized(ctx, tool.Name) { filtered = append(filtered, tool) } } @@ -202,23 +362,30 @@ func (r *Runtime) filterTools(_ context.Context, tools []mcp.Tool) []mcp.Tool { func (r *Runtime) handler(definition api.ToolDefinition) server.ToolHandlerFunc { return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) { - if _, ok := r.definitions[definition.Name]; !ok { + if !r.toolAuthorized(ctx, definition.Name) { + r.auditCall(ctx, definition.Name, "denied", "tool_not_delegated") return nil, fmt.Errorf("caller tool %q is not authorized", definition.Name) } callCtx, cancel := context.WithCancel(ctx) stop := context.AfterFunc(r.ctx, cancel) defer stop() defer cancel() + if capability, ok := ctx.Value(delegationContextKey{}).(*delegatedCapability); ok { + stopCapability := context.AfterFunc(capability.ctx, cancel) + defer stopCapability() + } input := request.GetArguments() if input == nil { input = map[string]any{} } toolUseID, generatedToolUseID, err := toolUseID(request, input) if err != nil { + r.auditCall(ctx, definition.Name, "denied", "invalid_tool_use_id") return mcp.NewToolResultError(err.Error()), nil } if definition.NeedsApproval() { if r.canUseTool == nil { + r.auditCall(ctx, definition.Name, "denied", "approval_broker_unavailable") return mcp.NewToolResultError("tool approval is required but no approval broker is configured"), nil } approvalCtx, approvalCancel := context.WithTimeout(callCtx, r.approvalTimeout) @@ -228,6 +395,7 @@ func (r *Runtime) handler(definition api.ToolDefinition) server.ToolHandlerFunc }) approvalCancel() if err != nil { + r.auditCall(ctx, definition.Name, "denied", "approval_failed") return mcp.NewToolResultError(err.Error()), nil } if !decision.Allow { @@ -235,6 +403,7 @@ func (r *Runtime) handler(definition api.ToolDefinition) server.ToolHandlerFunc if message == "" { message = "tool call denied" } + r.auditCall(ctx, definition.Name, "denied", "approval_denied") return mcp.NewToolResultError(message), nil } if decision.UpdatedInput != nil { @@ -242,24 +411,33 @@ func (r *Runtime) handler(definition api.ToolDefinition) server.ToolHandlerFunc } } if err := r.validateActive(callCtx); err != nil { + r.auditCall(ctx, definition.Name, "denied", "capability_inactive") return mcp.NewToolResultError(err.Error()), nil } if err := r.validateInput(definition.Name, input); err != nil { + r.auditCall(ctx, definition.Name, "denied", "invalid_input") + return mcp.NewToolResultError(err.Error()), nil + } + if err := r.validateActive(callCtx); err != nil { + r.auditCall(ctx, definition.Name, "denied", "capability_inactive") return mcp.NewToolResultError(err.Error()), nil } output, err := definition.Handler(callCtx, input) if err != nil { + r.auditCall(ctx, definition.Name, "error", "handler_failed") return mcp.NewToolResultError(err.Error()), nil } result, err := mcp.NewToolResultJSON(output) if err != nil { + r.auditCall(ctx, definition.Name, "error", "result_encoding_failed") return mcp.NewToolResultErrorf("marshal caller tool %q result: %v", definition.Name, err), nil } + r.auditCall(ctx, definition.Name, "allowed", "") return result, nil } } -func (r *Runtime) authorize(next http.Handler) http.Handler { +func (r *Runtime) authorizeLocal(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { host, _, err := net.SplitHostPort(request.RemoteAddr) if err != nil || !net.ParseIP(host).IsLoopback() { @@ -282,6 +460,37 @@ func (r *Runtime) authorize(next http.Handler) http.Handler { }) } +func (r *Runtime) guardToolName(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + if request.Method != http.MethodPost || request.Body == nil { + next.ServeHTTP(w, request) + return + } + body, err := io.ReadAll(io.LimitReader(request.Body, maxMCPRequestBytes+1)) + if err != nil || len(body) > maxMCPRequestBytes { + http.Error(w, "caller-tool MCP request is too large", http.StatusRequestEntityTooLarge) + return + } + _ = request.Body.Close() + request.Body = io.NopCloser(bytes.NewReader(body)) + request.ContentLength = int64(len(body)) + var call struct { + Method string `json:"method"` + Params struct { + Name string `json:"name"` + } `json:"params"` + } + if json.Unmarshal(body, &call) == nil && call.Method == "tools/call" { + if !r.toolAuthorized(request.Context(), call.Params.Name) { + r.auditCall(request.Context(), call.Params.Name, "denied", "tool_not_delegated") + http.Error(w, fmt.Sprintf("caller tool %q is not authorized", call.Params.Name), http.StatusForbidden) + return + } + } + next.ServeHTTP(w, request) + }) +} + func (r *Runtime) active() bool { return !r.revoked.Load() && (r.expiresAt.IsZero() || time.Now().Before(r.expiresAt)) } @@ -290,6 +499,9 @@ func (r *Runtime) validateActive(ctx context.Context) error { if !r.active() { return fmt.Errorf("caller-tool credential is inactive") } + if capability, ok := ctx.Value(delegationContextKey{}).(*delegatedCapability); ok && !capability.active() { + return fmt.Errorf("delegated caller-tool credential is inactive") + } if r.validate != nil { if err := r.validate(ctx); err != nil { return fmt.Errorf("validate caller-tool credential: %w", err) @@ -298,6 +510,122 @@ func (r *Runtime) validateActive(ctx context.Context) error { return nil } +func (r *Runtime) toolAuthorized(ctx context.Context, name string) bool { + if _, ok := r.definitions[name]; !ok { + return false + } + capability, ok := ctx.Value(delegationContextKey{}).(*delegatedCapability) + if !ok { + return true + } + _, ok = capability.tools[name] + return ok +} + +func remoteCapabilityID(path string) (string, bool) { + trimmed, ok := strings.CutPrefix(path, RemoteEndpointPrefix) + if !ok { + return "", false + } + id, suffix, ok := strings.Cut(trimmed, "/") + return id, ok && id != "" && suffix == strings.TrimPrefix(endpointPath, "/") +} + +func (capability *delegatedCapability) serveHTTP(w http.ResponseWriter, request *http.Request) { + if strings.TrimSpace(request.Header.Get("Origin")) != "" { + capability.runtime.auditEvent(capability.binding, AuditEvent{Action: "authentication", Result: "denied", Reason: "browser_origin"}) + http.Error(w, "caller-tool endpoint does not accept browser origins", http.StatusForbidden) + return + } + presented, ok := strings.CutPrefix(request.Header.Get("Authorization"), "Bearer ") + actual := sha256.Sum256([]byte(strings.TrimSpace(presented))) + if !ok || subtle.ConstantTimeCompare(actual[:], capability.tokenHash[:]) != 1 { + capability.runtime.auditEvent(capability.binding, AuditEvent{Action: "authentication", Result: "denied", Reason: "invalid_credential"}) + writeCredentialRejection(w) + return + } + if request.Header.Get(TaskHeader) != capability.binding.TaskID { + capability.runtime.auditEvent(capability.binding, AuditEvent{Action: "authentication", Result: "denied", Reason: "wrong_task"}) + http.Error(w, "caller-tool capability is not valid for this task", http.StatusForbidden) + return + } + if request.Header.Get(AgentHeader) != capability.binding.Agent { + capability.runtime.auditEvent(capability.binding, AuditEvent{Action: "authentication", Result: "denied", Reason: "wrong_agent"}) + http.Error(w, "caller-tool capability is not valid for this agent", http.StatusForbidden) + return + } + if !capability.active() || capability.runtime.validateActive(request.Context()) != nil { + capability.runtime.auditEvent(capability.binding, AuditEvent{Action: "authentication", Result: "denied", Reason: "inactive"}) + writeCredentialRejection(w) + return + } + forward := request.Clone(context.WithValue(request.Context(), delegationContextKey{}, capability)) + forward.URL = cloneURL(request.URL) + forward.URL.Path = endpointPath + forward.URL.RawPath = "" + capability.runtime.mcpHTTP.ServeHTTP(w, forward) +} + +func (capability *delegatedCapability) active() bool { + if capability.revoked.Load() || capability.expired.Load() { + return false + } + if time.Now().Before(capability.binding.ExpiresAt) { + return true + } + capability.expire() + return false +} + +func (capability *delegatedCapability) expire() { + if capability.revoked.Load() || !capability.expired.CompareAndSwap(false, true) { + return + } + remoteCapabilities.Lock() + if remoteCapabilities.values[capability.id] == capability { + delete(remoteCapabilities.values, capability.id) + } + remoteCapabilities.Unlock() + capability.cancel() + capability.runtime.auditEvent(capability.binding, AuditEvent{Action: "expiry", Result: "expired"}) +} + +func (capability *delegatedCapability) revoke(reason string) { + if capability.revoked.CompareAndSwap(false, true) { + capability.cancel() + capability.runtime.auditEvent(capability.binding, AuditEvent{Action: "revocation", Result: "revoked", Reason: reason}) + } +} + +func (r *Runtime) auditCall(ctx context.Context, tool, result, reason string) { + capability, _ := ctx.Value(delegationContextKey{}).(*delegatedCapability) + if capability == nil { + return + } + r.auditEvent(capability.binding, AuditEvent{Action: "call", Tool: tool, Result: result, Reason: reason}) +} + +func (r *Runtime) auditEvent(binding api.CallerToolBinding, event AuditEvent) { + event.TaskID = binding.TaskID + event.Agent = binding.Agent + if r.audit != nil { + r.audit(event) + return + } + callertoolsLog.Infof("caller-tool capability action=%s task=%s agent=%s tool=%s result=%s reason=%s", + event.Action, event.TaskID, event.Agent, event.Tool, event.Result, event.Reason) +} + +func writeCredentialRejection(w http.ResponseWriter) { + w.Header().Set("WWW-Authenticate", "Bearer") + http.Error(w, "invalid caller-tool credential", http.StatusUnauthorized) +} + +func cloneURL(value *url.URL) *url.URL { + cloned := *value + return &cloned +} + func mcpTool(definition api.ToolDefinition) (mcp.Tool, *jsonschema.Schema, error) { schema := definition.InputSchema if schema == nil { diff --git a/pkg/ai/client.go b/pkg/ai/client.go index 1fb579d4..8304e3b8 100644 --- a/pkg/ai/client.go +++ b/pkg/ai/client.go @@ -49,7 +49,17 @@ func NewProvider(cfg Config) (Provider, error) { } func newResolvedProvider(cfg Config) (Provider, error) { - p, err := api.NewProvider(cfg) + var p Provider + var err error + if cfg.SandboxSelection != nil { + if descriptor, ok := api.SandboxFor(cfg.SandboxSelection.Kind); ok && descriptor.Has(api.CapabilityRemoteExec) { + p, err = newRemoteProvider(cfg) + } else { + p, err = api.NewProvider(cfg) + } + } else { + p, err = api.NewProvider(cfg) + } if err != nil { return nil, err } diff --git a/pkg/ai/remote_provider.go b/pkg/ai/remote_provider.go new file mode 100644 index 00000000..2a52b364 --- /dev/null +++ b/pkg/ai/remote_provider.go @@ -0,0 +1,106 @@ +package ai + +import ( + "context" + "fmt" + "sync" + + "github.com/flanksource/captain/pkg/api" +) + +// remoteProvider adapts a whole-run remote sandbox to the provider contract. +// Runtime-only caller-tool authority is attached to the sandbox selection and +// never projected into the serializable request. +type remoteProvider struct { + executor api.RemoteExecutor + sandbox api.Sandbox + model string + backend api.Backend + tools bool + + prepareOnce sync.Once + prepareErr error + closeOnce sync.Once + closeErr error +} + +func newRemoteProvider(cfg Config) (Provider, error) { + selection := *cfg.SandboxSelection + descriptor, ok := api.SandboxFor(selection.Kind) + if !ok { + return nil, fmt.Errorf("unknown sandbox kind %q", selection.Kind) + } + if err := descriptor.ValidateMode(cfg.Model.Backend.Mode()); err != nil { + return nil, err + } + if len(selection.CallerTools) > 0 && cfg.CallerTools == nil { + return nil, fmt.Errorf("delegated caller tools require a supervisor caller-tool endpoint") + } + if len(selection.CallerTools) > 0 && !api.SupportsCallerTools(cfg.Model.Backend) { + return nil, fmt.Errorf("remote backend %q does not support delegated caller tools", cfg.Model.Backend) + } + selection.CallerToolEndpoint = cfg.CallerTools + sandbox, err := api.NewSandbox(selection) + if err != nil { + return nil, err + } + executor, ok := api.SandboxAs[api.RemoteExecutor](sandbox) + if !ok { + _ = sandbox.Close() + return nil, fmt.Errorf("sandbox %q declares remote execution but provides none", selection.Kind) + } + return &remoteProvider{ + executor: executor, sandbox: sandbox, model: cfg.Model.Name, + backend: cfg.Model.Backend, tools: cfg.CallerTools != nil, + }, nil +} + +func (provider *remoteProvider) Execute(ctx context.Context, request Request) (*Response, error) { + provider.prepareOnce.Do(func() { + _, provider.prepareErr = provider.sandbox.Prepare(ctx, &request) + }) + if provider.prepareErr != nil { + return nil, provider.prepareErr + } + return provider.executor.Execute(ctx, request) +} + +func (provider *remoteProvider) ExecuteStream(ctx context.Context, request Request) (<-chan Event, error) { + events := make(chan Event, 3) + go func() { + defer close(events) + response, err := provider.Execute(ctx, request) + if err != nil { + emitRemoteEvent(ctx, events, Event{Kind: EventError, Error: err.Error(), Model: provider.model}) + emitRemoteEvent(ctx, events, Event{Kind: EventResult, Success: false, Error: err.Error(), Model: provider.model}) + return + } + if response.Text != "" { + emitRemoteEvent(ctx, events, Event{Kind: EventText, Text: response.Text, Model: provider.model}) + } + emitRemoteEvent(ctx, events, Event{ + Kind: EventResult, Success: true, Model: provider.model, + Usage: &response.Usage, CostUSD: response.CostUSD, + }) + }() + return events, nil +} + +func emitRemoteEvent(ctx context.Context, events chan<- Event, event Event) { + select { + case events <- event: + case <-ctx.Done(): + } +} + +func (provider *remoteProvider) GetModel() string { return provider.model } +func (provider *remoteProvider) GetBackend() api.Backend { return provider.backend } + +func (provider *remoteProvider) SupportsCallerTools() bool { return provider.tools } + +func (provider *remoteProvider) Close() error { + provider.closeOnce.Do(func() { + provider.closeErr = provider.sandbox.Close() + }) + return provider.closeErr +} diff --git a/pkg/aichat/service.go b/pkg/aichat/service.go index 826a9d88..99bb0101 100644 --- a/pkg/aichat/service.go +++ b/pkg/aichat/service.go @@ -231,6 +231,13 @@ func (s *Service) handleChat(w http.ResponseWriter, request *http.Request) { config.SessionID = spec.SessionID config.CaptainSessionID = chat.ThreadID config.Tools = definitions + if config.SandboxSelection != nil && spec.Sandbox != nil { + selection := *config.SandboxSelection + selection.Agent = spec.Sandbox.Agent + selection.CallerTools = append([]string(nil), spec.Sandbox.CallerTools...) + selection.Policy = spec.Sandbox.Policy + config.SandboxSelection = &selection + } config, err = s.prepareProviderConfig(request.Context(), config) if err != nil { http.Error(w, err.Error(), http.StatusServiceUnavailable) diff --git a/pkg/api/runtime_config.go b/pkg/api/runtime_config.go index e63fc7c0..e35b0024 100644 --- a/pkg/api/runtime_config.go +++ b/pkg/api/runtime_config.go @@ -48,8 +48,35 @@ type CallerToolEndpoint struct { Name string URL string Headers map[string]string + // Delegate issues a narrower transport capability for one remote task. It + // is present only on Captain-owned endpoints and never crosses a wire. + Delegate CallerToolDelegateFunc `json:"-"` } +// CallerToolBinding is the identity and lifetime a delegated endpoint must +// re-check independently of its bearer credential. +type CallerToolBinding struct { + TaskID string + Agent string + ExpiresAt time.Time + // ToolNames is the requested child allowlist. Delegation intersects it with + // the already-resolved parent definitions, so unknown or parent-denied names + // can never acquire authority. + ToolNames []string +} + +// CallerToolDelegation is a task-scoped endpoint plus its revocation hook. +// The endpoint headers contain a plaintext capability and must stay in memory +// or an authenticated secret-delivery channel. +type CallerToolDelegation struct { + Endpoint CallerToolEndpoint + Revoke func() +} + +// CallerToolDelegateFunc issues a task-scoped child of an already-resolved +// caller-tool endpoint. The child can retain but never widen the parent's tools. +type CallerToolDelegateFunc func(context.Context, CallerToolBinding) (*CallerToolDelegation, error) + func (endpoint CallerToolEndpoint) Validate() error { if endpoint.Name == "" { return fmt.Errorf("caller-tool endpoint name is required") diff --git a/pkg/api/sandbox_ref.go b/pkg/api/sandbox_ref.go index abbf0c74..94acc14a 100644 --- a/pkg/api/sandbox_ref.go +++ b/pkg/api/sandbox_ref.go @@ -17,6 +17,7 @@ import ( // sandbox: # object: backend plus overrides // backend: prod-pool // agent: worker-01 +// callerTools: [invoice_get, invoice_update] // policy: {paths: ["pkg/**"], maxAttempts: 3} // // The scalar form is sugar for {backend: }. Whether the name is a bare @@ -28,6 +29,9 @@ type SandboxRef struct { Backend string `json:"backend,omitempty" yaml:"backend,omitempty"` // Agent optionally pins one enrolled agent of a git-agent backend. Agent string `json:"agent,omitempty" yaml:"agent,omitempty"` + // CallerTools is the exact caller-tool allowlist delegated to a remote run. + // Omitted means no caller tools leave the supervisor. + CallerTools []string `json:"callerTools,omitempty" yaml:"callerTools,omitempty"` // Policy optionally overrides the backend's dispatch policy for this run. Policy *SandboxPolicy `json:"policy,omitempty" yaml:"policy,omitempty"` } @@ -48,7 +52,7 @@ type sandboxRefAlias SandboxRef // isScalar reports whether the ref carries only a backend name and so can // round-trip through the scalar form. func (r SandboxRef) isScalar() bool { - return r.Agent == "" && r.Policy == nil + return r.Agent == "" && len(r.CallerTools) == 0 && r.Policy == nil } func (r SandboxRef) MarshalJSON() ([]byte, error) { @@ -116,13 +120,17 @@ func (r *SandboxRef) UnmarshalYAML(value *yaml.Node) error { if err := val.Decode(&r.Agent); err != nil { return err } + case "callerTools": + if err := val.Decode(&r.CallerTools); err != nil { + return err + } case "policy": r.Policy = &SandboxPolicy{} if err := r.Policy.unmarshalStrict(val); err != nil { return err } default: - return fmt.Errorf("unknown sandbox key %q (valid: backend, agent, policy)", key) + return fmt.Errorf("unknown sandbox key %q (valid: backend, agent, callerTools, policy)", key) } } return nil @@ -165,6 +173,8 @@ func (SandboxRef) JSONSchema() *jsonschema.Schema { Description: "Configured sandbox backend from ~/.captain.yaml, or a bare adapter kind"}) properties.Set("agent", &jsonschema.Schema{Type: "string", Description: "Pin one enrolled agent of a git-agent backend"}) + properties.Set("callerTools", &jsonschema.Schema{Type: "array", Items: &jsonschema.Schema{Type: "string"}, + Description: "Exact caller-tool names delegated to the remote run; omitted delegates none"}) properties.Set("policy", &jsonschema.Schema{Type: "object", Properties: policyProperties, AdditionalProperties: jsonschema.FalseSchema}) return &jsonschema.Schema{ @@ -184,10 +194,20 @@ func (SandboxRef) JSONSchema() *jsonschema.Schema { func (r SandboxRef) Validate() error { if r.Backend == "" { if !r.isScalar() { - return fmt.Errorf("sandbox overrides (agent/policy) require a backend") + return fmt.Errorf("sandbox overrides (agent/callerTools/policy) require a backend") } return fmt.Errorf("sandbox must name a backend or adapter kind (one of: %s)", SandboxKindList()) } + seen := make(map[string]struct{}, len(r.CallerTools)) + for i, name := range r.CallerTools { + if name == "" { + return fmt.Errorf("sandbox callerTools entry %d cannot be empty", i) + } + if _, exists := seen[name]; exists { + return fmt.Errorf("sandbox callerTools contains duplicate %q", name) + } + seen[name] = struct{}{} + } if r.Policy != nil && r.Policy.MaxAttempts < 0 { return fmt.Errorf("sandbox policy maxAttempts must be >= 0, got %d", r.Policy.MaxAttempts) } diff --git a/pkg/api/sandbox_registry.go b/pkg/api/sandbox_registry.go index 53bd4364..16a4cf1d 100644 --- a/pkg/api/sandbox_registry.go +++ b/pkg/api/sandbox_registry.go @@ -22,8 +22,14 @@ type SandboxConfig struct { // Agent pins one enrolled agent of a git-agent backend, from // SandboxRef.Agent. Empty lets the adapter choose. Agent string `json:"agent,omitempty" yaml:"agent,omitempty"` + // CallerTools is the exact per-run remote delegation allowlist copied from + // SandboxRef. An empty list preserves the no-delegation behavior. + CallerTools []string `json:"callerTools,omitempty" yaml:"callerTools,omitempty"` // Policy is the per-run override from SandboxRef.Policy. Policy *SandboxPolicy `json:"policy,omitempty" yaml:"policy,omitempty"` + // CallerToolEndpoint is runtime-only authority handed to a remote executor. It is + // excluded from resolved specs because its headers contain a live capability. + CallerToolEndpoint *CallerToolEndpoint `json:"-" yaml:"-"` } // Options keys shared between the git-agent hook resolver and the adapters it diff --git a/pkg/cli/ai_prompt_file.go b/pkg/cli/ai_prompt_file.go index 099c8d36..cdfae627 100644 --- a/pkg/cli/ai_prompt_file.go +++ b/pkg/cli/ai_prompt_file.go @@ -196,6 +196,7 @@ func overlayCLI(base ai.Request, baseCfg ai.Config, o AIPromptOptions) (ai.Reque ref := api.SandboxRef{Backend: selector} if base.Sandbox != nil { ref.Agent = base.Sandbox.Agent + ref.CallerTools = base.Sandbox.CallerTools ref.Policy = base.Sandbox.Policy } req.Sandbox = &ref diff --git a/pkg/cli/ai_sandbox.go b/pkg/cli/ai_sandbox.go index 89029dc9..71661b7c 100644 --- a/pkg/cli/ai_sandbox.go +++ b/pkg/cli/ai_sandbox.go @@ -51,6 +51,7 @@ func recordSandboxSelection(req *ai.Request, cfg *ai.Config, selection captainco ref := api.SandboxRef{Backend: flagSelector} if req.Sandbox != nil { ref.Agent = req.Sandbox.Agent + ref.CallerTools = req.Sandbox.CallerTools ref.Policy = req.Sandbox.Policy } req.Sandbox = &ref @@ -99,6 +100,7 @@ func sandboxSelectionConfig(selection captainconfig.SandboxSelection, ref *api.S cfg := &api.SandboxConfig{Kind: selection.Kind, Name: selection.Name, Options: selection.Options} if ref != nil { cfg.Agent = ref.Agent + cfg.CallerTools = ref.CallerTools cfg.Policy = ref.Policy } return cfg diff --git a/pkg/cli/ai_sandbox_remote.go b/pkg/cli/ai_sandbox_remote.go index 320ee947..c443224c 100644 --- a/pkg/cli/ai_sandbox_remote.go +++ b/pkg/cli/ai_sandbox_remote.go @@ -49,8 +49,9 @@ func remoteExecProviderFor(req *ai.Request, cfg ai.Config) (ai.Provider, error) if !relocatesRun(cfg) { return nil, nil } - selection := cfg.SandboxSelection - sandbox, err := api.NewSandbox(*selection) + selection := *cfg.SandboxSelection + selection.CallerToolEndpoint = cfg.CallerTools + sandbox, err := api.NewSandbox(selection) if err != nil { return nil, err } diff --git a/pkg/cli/gitagent_hook.go b/pkg/cli/gitagent_hook.go index 7a23e92a..22ad1520 100644 --- a/pkg/cli/gitagent_hook.go +++ b/pkg/cli/gitagent_hook.go @@ -146,6 +146,7 @@ func hookRuntimeFromConfig(backendName string) (gitagent.HookRuntime, error) { url, _ := supervisor["url"].(string) hostFP, _ := supervisor["hostFingerprint"].(string) tokenPath, _ := supervisor["tokenPath"].(string) + rt.Agent, _ = supervisor["agent"].(string) keysDir, err := gitAgentKeysDir() if err != nil { return rt, err diff --git a/pkg/cli/gitagent_runtask.go b/pkg/cli/gitagent_runtask.go index 9aac6b9c..9687ebac 100644 --- a/pkg/cli/gitagent_runtask.go +++ b/pkg/cli/gitagent_runtask.go @@ -61,9 +61,19 @@ func RunGitAgentRunTask(ctx context.Context, opts GitAgentRunTaskOptions) (_ any if err != nil { return nil, err } + var callerTools *api.CallerToolEndpoint + if payload.CallerTools { + if !api.SupportsCallerTools(api.Backend(payload.Backend)) { + return nil, fmt.Errorf("remote backend %q does not support delegated caller tools", payload.Backend) + } + callerTools, err = gitagent.LoadCallerToolEndpoint(opts.Repo, opts.Task) + if err != nil { + return nil, fmt.Errorf("load delegated caller tools: %w", err) + } + } identity := ai.LogIdentity(api.Backend(payload.Backend), payload.Model, payload.Effort) log.Infof("git-agent task %s starting %s in %s", opts.Task, identity, worktree) - if err := runTaskPrompt(ctx, worktree, payload); err != nil { + if err := runTaskPrompt(ctx, worktree, payload, callerTools); err != nil { return nil, fmt.Errorf("running the dispatched prompt: %w", err) } log.Infof("git-agent task %s agent finished after %s; preparing submission", opts.Task, time.Since(started).Round(time.Millisecond)) @@ -79,7 +89,16 @@ func RunGitAgentRunTask(ctx context.Context, opts GitAgentRunTaskOptions) (_ any // runTaskPrompt executes the dispatched prompt in the worktree. The sandbox is // pinned to none: this process IS the relocated run, so resolving a relocating // sandbox here would dispatch the task to another agent, and so on (H15). -func runTaskPrompt(ctx context.Context, worktree string, payload gitagent.TaskPayload) error { +func runTaskPrompt( + ctx context.Context, + worktree string, + payload gitagent.TaskPayload, + callerToolEndpoints ...*api.CallerToolEndpoint, +) error { + var callerTools *api.CallerToolEndpoint + if len(callerToolEndpoints) > 0 { + callerTools = callerToolEndpoints[0] + } providerOpts := AIProviderOptions{ ModelFlags: aiflags.ModelFlags{Model: payload.Model, Backend: payload.Backend, Effort: string(payload.Effort)}, Sandbox: "none", @@ -88,6 +107,7 @@ func runTaskPrompt(ctx context.Context, worktree string, payload gitagent.TaskPa if err != nil { return err } + cfg.CallerTools = callerTools var req ai.Request req.Prompt.User = payload.Prompt req.Prompt.System = payload.System diff --git a/pkg/cli/gitagent_serve.go b/pkg/cli/gitagent_serve.go index 3ccb015e..95559340 100644 --- a/pkg/cli/gitagent_serve.go +++ b/pkg/cli/gitagent_serve.go @@ -87,7 +87,7 @@ func RunGitAgentServe(ctx context.Context, opts GitAgentServeOptions) (any, erro if transport == transportHTTPS { startSidecarBackground(ctx, root) return nil, serveSidecarHTTPS(ctx, sidecarHTTPSPlan{ - listen: opts.Listen, root: root, keysDir: keysDir, advertise: opts.Advertise, + listen: opts.Listen, root: root, keysDir: keysDir, advertise: opts.Advertise, backend: opts.Backend, certPath: opts.TLSCert, keyPath: opts.TLSKey, }) } diff --git a/pkg/cli/gitagent_serve_https.go b/pkg/cli/gitagent_serve_https.go index 680fbff7..d8b003e0 100644 --- a/pkg/cli/gitagent_serve_https.go +++ b/pkg/cli/gitagent_serve_https.go @@ -16,6 +16,7 @@ import ( "crypto/tls" "encoding/json" "fmt" + "net" "net/http" "net/url" "path/filepath" @@ -34,6 +35,7 @@ type sidecarHTTPSPlan struct { root string keysDir string advertise string + backend string certPath string keyPath string } @@ -54,6 +56,10 @@ func serveSidecarHTTPS(ctx context.Context, plan sidecarHTTPSPlan) error { "rerun with --token-file to enroll", err) } identify := sidecarIdentity(dispatch) + runtime, err := hookRuntimeFromConfig(plan.backend) + if err != nil { + return err + } handler, err := gitagent.NewHTTPHandler(gitagent.HTTPServerConfig{ Root: plan.root, Role: gitagent.RoleSidecar, @@ -69,6 +75,45 @@ func serveSidecarHTTPS(ctx context.Context, plan sidecarHTTPSPlan) error { mux := http.NewServeMux() mux.Handle(gitagent.GitHTTPPrefix, handler) mux.Handle("POST "+gitagent.AgentWhoamiPath, agentWhoamiHandler(identify, RunWhoami)) + callerTools := unsupportedCallerToolProxy(identify, + "delegated caller tools require this sidecar to enroll with an HTTPS supervisor") + var callerToolServer *http.Server + if gitagent.EndpointScheme(runtime.Relay.URL) == "https" && strings.TrimSpace(runtime.Agent) == "" { + callerTools = unsupportedCallerToolProxy(identify, + "delegated caller tools require re-enrollment so this sidecar has a bound agent identity") + } else if gitagent.EndpointScheme(runtime.Relay.URL) == "https" { + // The model reaches its co-located sidecar over loopback HTTP, while the + // cross-host proxy leg stays pinned HTTPS. This avoids teaching every MCP + // client how to trust the sidecar's default self-signed Git certificate. + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return fmt.Errorf("listen for delegated caller tools: %w", err) + } + callerTools, err = gitagent.NewCallerToolProxy(gitagent.CallerToolProxyConfig{ + Root: filepath.Join(plan.root, SidecarRepoName), EndpointURL: "http://" + listener.Addr().String(), + SupervisorURL: runtime.Relay.URL, SupervisorCAPath: runtime.Relay.CAPath, + SupervisorPublicKey: runtime.Relay.PinnedPublicKey, + Agent: runtime.Agent, + DefaultRunner: strings.TrimSpace(runtime.AgentCommand) == "", + IdentifySupervisor: identify, Log: log.Infof, + }) + if err != nil { + _ = listener.Close() + return err + } + callerToolMux := http.NewServeMux() + callerToolMux.Handle(gitagent.CallerToolPath+"/", callerTools) + callerToolServer = &http.Server{ + Handler: callerToolMux, ReadHeaderTimeout: 5 * time.Second, + } + go func() { + if err := callerToolServer.Serve(listener); err != nil && err != http.ErrServerClosed { + log.Errorf("delegated caller-tool listener stopped: %v", err) + } + }() + } + mux.Handle(gitagent.CallerToolPath, callerTools) + mux.Handle(gitagent.CallerToolPath+"/", callerTools) server := &http.Server{ Addr: plan.listen, @@ -87,6 +132,14 @@ func serveSidecarHTTPS(ctx context.Context, plan sidecarHTTPSPlan) error { go func() { <-ctx.Done() _ = server.Close() + if callerToolServer != nil { + _ = callerToolServer.Close() + } + }() + defer func() { + if callerToolServer != nil { + _ = callerToolServer.Close() + } }() if err := server.ListenAndServeTLS("", ""); err != nil && ctx.Err() == nil { return err @@ -94,6 +147,19 @@ func serveSidecarHTTPS(ctx context.Context, plan sidecarHTTPSPlan) error { return nil } +func unsupportedCallerToolProxy( + identify func(*http.Request) (string, error), + reason string, +) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + if _, err := identify(request); err != nil { + http.Error(w, "caller-tool control requires the enrolled supervisor credential", http.StatusForbidden) + return + } + http.Error(w, reason, http.StatusConflict) + }) +} + func sidecarTLSConfig(plan sidecarHTTPSPlan, host string) (*gitagent.TLSCredential, *tls.Config, error) { certPath, keyPath := strings.TrimSpace(plan.certPath), strings.TrimSpace(plan.keyPath) if (certPath == "") != (keyPath == "") { diff --git a/pkg/cli/serve.go b/pkg/cli/serve.go index 0c779ab3..ee5ba206 100644 --- a/pkg/cli/serve.go +++ b/pkg/cli/serve.go @@ -19,6 +19,7 @@ import ( "syscall" "time" + "github.com/flanksource/captain/pkg/ai/callertools" "github.com/flanksource/captain/pkg/aichat" "github.com/flanksource/captain/pkg/api" "github.com/flanksource/captain/pkg/captaintoken" @@ -240,6 +241,8 @@ func RunServe(ctx context.Context, rootCmd *cobra.Command, opts ServeOptions, ve root := http.NewServeMux() root.Handle("/api/", mux) root.Handle("/health", mux) + root.Handle(gitagent.GitHTTPPrefix, mux) + root.Handle(callertools.RemoteEndpointPrefix, callertools.RemoteHandler()) root.Handle("/", uiHandler) tlsConfig := serveTLSConfig(certificate) diff --git a/pkg/cli/serve_auth.go b/pkg/cli/serve_auth.go index 6e6d0faf..575b5ee0 100644 --- a/pkg/cli/serve_auth.go +++ b/pkg/cli/serve_auth.go @@ -28,7 +28,7 @@ const gitPathPrefix = gitagent.GitHTTPPrefix type tokenContextKey struct{} // TokenFromContext returns the credential a request authenticated with. ok is -// false for a loopback request, which carries none — a handler that needs an +// false for an unauthenticated loopback request — a handler that needs an // identity must say so rather than assume one. func TokenFromContext(ctx context.Context) (captaintoken.Record, bool) { record, ok := ctx.Value(tokenContextKey{}).(captaintoken.Record) @@ -47,18 +47,18 @@ type TokenAuthConfig struct { // TokenAuthMiddleware requires a captain token for requests that arrive from // off this machine. // -// Loopback is exempt, so the local webapp, CLI and hook subprocesses are -// untouched. That is not just convenience: an EventSource stream cannot set an -// Authorization header, so requiring one would break the UI for the ordinary -// local case. The exemption rests entirely on RemoteAddr, which is why a -// request carrying proxy forwarding headers is treated as remote wherever it -// connected from — otherwise anything behind a same-host reverse proxy would -// inherit it. +// Loopback requests without a credential are exempt, so the local webapp, CLI +// and hook subprocesses are untouched. A loopback caller that presents a bearer +// is verified so identity-bearing protocols such as Git-agent enrollment work +// when both processes share a host. The exemption rests entirely on RemoteAddr, +// which is why a request carrying proxy forwarding headers is treated as remote +// wherever it connected from — otherwise anything behind a same-host reverse +// proxy would inherit it. func TokenAuthMiddleware(config TokenAuthConfig) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { scope, protected := requiredScope(r.URL.Path) - if !protected || isLoopbackRequest(r) { + if !protected || (isLoopbackRequest(r) && strings.TrimSpace(r.Header.Get("Authorization")) == "") { next.ServeHTTP(w, r) return } diff --git a/pkg/gitagent/callertools.go b/pkg/gitagent/callertools.go new file mode 100644 index 00000000..9c7e2cf6 --- /dev/null +++ b/pkg/gitagent/callertools.go @@ -0,0 +1,581 @@ +// Caller-tool talkback keeps task capabilities out of the Git protocol. The +// authenticated supervisor delivers one grant to an HTTPS sidecar, which +// briefly materializes it as a task-local secret and proxies MCP requests back +// to the supervisor without ever using the durable Git transport token as tool +// authority. +package gitagent + +import ( + "bytes" + "context" + "crypto/sha256" + "crypto/subtle" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/flanksource/captain/pkg/ai/callertools" + "github.com/flanksource/captain/pkg/api" +) + +const ( + CallerToolPath = "/api/v1/caller-tools" + callerToolSecretFile = "caller-tools.json" + maxCallerToolGrantBytes = 64 << 10 +) + +// CallerToolGrant is delivered over the sidecar's authenticated HTTPS control +// path. Endpoint headers are plaintext capability material and are never JSON +// marshaled directly by callers. +type CallerToolGrant struct { + Task string + Agent string + Endpoint api.CallerToolEndpoint `json:"-"` + ExpiresAt time.Time +} + +type callerToolGrantWire struct { + Task string `json:"task"` + Agent string `json:"agent"` + Name string `json:"name"` + Route string `json:"route"` + Credential string `json:"credential"` + ExpiresAt time.Time `json:"expiresAt"` +} + +type callerToolEndpointSecret struct { + Name string `json:"name"` + URL string `json:"url"` + Headers map[string]string `json:"headers"` +} + +// CallerToolProxyConfig binds a sidecar proxy to its enrolled supervisor and +// standard Captain task runner. +type CallerToolProxyConfig struct { + Root string + EndpointURL string + SupervisorURL string + SupervisorCAPath string + SupervisorPublicKey string + Agent string + DefaultRunner bool + IdentifySupervisor func(*http.Request) (string, error) + Log func(format string, args ...any) +} + +type callerToolProxy struct { + root string + endpointURL *url.URL + supervisorURL *url.URL + client *http.Client + agent string + defaultRunner bool + identifySupervisor func(*http.Request) (string, error) + log func(format string, args ...any) + + mu sync.RWMutex + sessions map[string]*callerToolSession +} + +type callerToolSession struct { + task string + agent string + route string + tokenHash [sha256.Size]byte + expiresAt time.Time + revoked atomic.Bool +} + +// NewCallerToolProxy creates the sidecar handler used for authenticated grant +// delivery and task-scoped MCP proxying. +func NewCallerToolProxy(config CallerToolProxyConfig) (http.Handler, error) { + if strings.TrimSpace(config.Root) == "" || strings.TrimSpace(config.Agent) == "" || config.IdentifySupervisor == nil { + return nil, fmt.Errorf("caller-tool proxy requires a sidecar root, enrolled agent, and supervisor identity resolver") + } + endpoint, err := parseCallerToolBase(config.EndpointURL) + if err != nil { + return nil, err + } + supervisor, err := parseHTTPSBase(config.SupervisorURL, "enrolled supervisor URL") + if err != nil { + return nil, err + } + client, err := HTTPSClient(config.SupervisorCAPath, config.SupervisorPublicKey) + if err != nil { + return nil, err + } + logf := config.Log + if logf == nil { + logf = func(string, ...any) {} + } + removed, err := removeStaleCallerToolSecrets(config.Root) + if err != nil { + return nil, err + } + if removed > 0 { + logf("git-agent caller-tool startup removed %d stale task secret(s)", removed) + } + proxy := &callerToolProxy{ + root: config.Root, endpointURL: endpoint, supervisorURL: supervisor, + client: client, agent: config.Agent, defaultRunner: config.DefaultRunner, + identifySupervisor: config.IdentifySupervisor, log: logf, + sessions: map[string]*callerToolSession{}, + } + return http.HandlerFunc(proxy.serveHTTP), nil +} + +func (proxy *callerToolProxy) serveHTTP(w http.ResponseWriter, request *http.Request) { + if request.URL.Path == CallerToolPath { + if request.Method == http.MethodPost { + proxy.register(w, request) + return + } + http.NotFound(w, request) + return + } + trimmed, ok := strings.CutPrefix(request.URL.Path, CallerToolPath+"/") + if !ok { + http.NotFound(w, request) + return + } + parts := strings.Split(trimmed, "/") + switch { + case len(parts) == 2 && parts[1] == "mcp": + proxy.forward(w, request, parts[0]) + case len(parts) == 1 && request.Method == http.MethodDelete: + proxy.revoke(w, request, parts[0]) + default: + http.NotFound(w, request) + } +} + +func (proxy *callerToolProxy) register(w http.ResponseWriter, request *http.Request) { + if _, err := proxy.identifySupervisor(request); err != nil { + http.Error(w, "caller-tool grant requires the enrolled supervisor credential", http.StatusForbidden) + return + } + if !proxy.defaultRunner { + http.Error(w, "delegated caller tools require Captain's default git-agent task runner", http.StatusConflict) + return + } + payload, err := io.ReadAll(io.LimitReader(request.Body, maxCallerToolGrantBytes+1)) + if err != nil || len(payload) > maxCallerToolGrantBytes { + http.Error(w, "caller-tool grant is too large", http.StatusRequestEntityTooLarge) + return + } + var grant callerToolGrantWire + if err := json.Unmarshal(payload, &grant); err != nil { + http.Error(w, "invalid caller-tool grant", http.StatusBadRequest) + return + } + if err := validateCallerToolGrant(grant); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if grant.Agent != proxy.agent { + http.Error(w, "caller-tool grant is bound to a different agent", http.StatusForbidden) + return + } + endpointURL := *proxy.endpointURL + endpointURL.Path = CallerToolPath + "/" + grant.Task + "/mcp" + endpointURL.RawPath, endpointURL.RawQuery, endpointURL.Fragment = "", "", "" + secret := callerToolEndpointSecret{ + Name: grant.Name, URL: endpointURL.String(), + Headers: map[string]string{ + "Authorization": "Bearer " + grant.Credential, + callertools.TaskHeader: grant.Task, + callertools.AgentHeader: grant.Agent, + }, + } + session := &callerToolSession{ + task: grant.Task, agent: grant.Agent, route: grant.Route, + tokenHash: sha256.Sum256([]byte(grant.Credential)), expiresAt: grant.ExpiresAt, + } + proxy.mu.Lock() + if existing := proxy.sessions[grant.Task]; existing != nil && !existing.revoked.Load() { + proxy.mu.Unlock() + http.Error(w, "caller-tool grant already exists for task", http.StatusConflict) + return + } + if err := writeCallerToolSecret(proxy.root, grant.Task, secret); err != nil { + proxy.mu.Unlock() + http.Error(w, "store caller-tool task secret", http.StatusInternalServerError) + return + } + proxy.sessions[grant.Task] = session + proxy.mu.Unlock() + time.AfterFunc(time.Until(grant.ExpiresAt), func() { proxy.expire(session) }) + proxy.log("git-agent caller-tool grant issued task=%s agent=%s expires=%s", + grant.Task, grant.Agent, grant.ExpiresAt.UTC().Format(time.RFC3339)) + w.WriteHeader(http.StatusNoContent) +} + +func (proxy *callerToolProxy) forward(w http.ResponseWriter, request *http.Request, task string) { + if ValidateTaskID(task) != nil { + writeCallerToolRejection(w) + return + } + proxy.mu.RLock() + session := proxy.sessions[task] + proxy.mu.RUnlock() + if session == nil || session.revoked.Load() { + writeCallerToolRejection(w) + return + } + if !time.Now().Before(session.expiresAt) { + proxy.expire(session) + writeCallerToolRejection(w) + return + } + if strings.TrimSpace(request.Header.Get("Origin")) != "" { + http.Error(w, "caller-tool endpoint does not accept browser origins", http.StatusForbidden) + return + } + presented, ok := strings.CutPrefix(request.Header.Get("Authorization"), "Bearer ") + hash := sha256.Sum256([]byte(strings.TrimSpace(presented))) + if !ok || subtle.ConstantTimeCompare(hash[:], session.tokenHash[:]) != 1 { + proxy.log("git-agent caller-tool request denied task=%s agent=%s reason=invalid_credential", session.task, session.agent) + writeCallerToolRejection(w) + return + } + if request.Header.Get(callertools.TaskHeader) != session.task || + request.Header.Get(callertools.AgentHeader) != session.agent { + proxy.log("git-agent caller-tool request denied task=%s agent=%s reason=binding_mismatch", session.task, session.agent) + http.Error(w, "caller-tool capability binding does not match this task", http.StatusForbidden) + return + } + upstreamURL := *proxy.supervisorURL + upstreamURL.Path = session.route + upstreamURL.RawPath, upstreamURL.RawQuery, upstreamURL.Fragment = "", request.URL.RawQuery, "" + upstream := request.Clone(request.Context()) + upstream.URL = &upstreamURL + upstream.RequestURI = "" + upstream.Host = upstreamURL.Host + removeHopByHopHeaders(upstream.Header) + upstream.Header.Set(callertools.TaskHeader, session.task) + upstream.Header.Set(callertools.AgentHeader, session.agent) + response, err := proxy.client.Do(upstream) + if err != nil { + proxy.log("git-agent caller-tool proxy failed task=%s agent=%s", session.task, session.agent) + http.Error(w, "caller-tool supervisor is unavailable", http.StatusBadGateway) + return + } + defer response.Body.Close() + copyHTTPHeaders(w.Header(), response.Header) + w.WriteHeader(response.StatusCode) + _, _ = io.Copy(w, response.Body) +} + +func (proxy *callerToolProxy) revoke(w http.ResponseWriter, request *http.Request, task string) { + if _, err := proxy.identifySupervisor(request); err != nil { + http.Error(w, "caller-tool revocation requires the enrolled supervisor credential", http.StatusForbidden) + return + } + if err := ValidateTaskID(task); err != nil { + http.NotFound(w, request) + return + } + proxy.mu.Lock() + session := proxy.sessions[task] + delete(proxy.sessions, task) + proxy.mu.Unlock() + if session != nil && session.revoked.CompareAndSwap(false, true) { + proxy.log("git-agent caller-tool grant revoked task=%s agent=%s", session.task, session.agent) + } + _ = removeCallerToolSecret(proxy.root, task) + w.WriteHeader(http.StatusNoContent) +} + +func (proxy *callerToolProxy) expire(session *callerToolSession) { + if session.revoked.CompareAndSwap(false, true) { + proxy.mu.Lock() + if proxy.sessions[session.task] == session { + delete(proxy.sessions, session.task) + } + proxy.mu.Unlock() + proxy.log("git-agent caller-tool grant expired task=%s agent=%s", session.task, session.agent) + _ = removeCallerToolSecret(proxy.root, session.task) + } +} + +// RegisterCallerTools delivers a task capability using the durable dispatch +// credential only to authenticate the delivery channel. +func RegisterCallerTools(ctx context.Context, target TransportTarget, grant CallerToolGrant) error { + wire, err := callerToolGrantToWire(grant) + if err != nil { + return err + } + body, err := json.Marshal(wire) + if err != nil { + return err + } + request, err := newCallerToolControlRequest(ctx, http.MethodPost, target, "", bytes.NewReader(body)) + if err != nil { + return err + } + request.Header.Set("Content-Type", "application/json") + client, err := HTTPSClient(target.CAPath, target.PinnedPublicKey) + if err != nil { + return err + } + defer client.CloseIdleConnections() + return doCallerToolControl(client, request) +} + +// RevokeCallerTools removes any unconsumed task secret and disables the +// sidecar proxy session. Supervisor-side revocation remains authoritative if +// this best-effort cleanup cannot reach the sidecar. +func RevokeCallerTools(ctx context.Context, target TransportTarget, task string) error { + request, err := newCallerToolControlRequest(ctx, http.MethodDelete, target, "/"+task, nil) + if err != nil { + return err + } + client, err := HTTPSClient(target.CAPath, target.PinnedPublicKey) + if err != nil { + return err + } + defer client.CloseIdleConnections() + return doCallerToolControl(client, request) +} + +// LoadCallerToolEndpoint consumes the sidecar-delivered secret before the +// remote model process starts. +func LoadCallerToolEndpoint(sidecarRepo, task string) (*api.CallerToolEndpoint, error) { + if err := ValidateTaskID(task); err != nil { + return nil, err + } + path := callerToolSecretPath(sidecarRepo, task) + info, err := os.Stat(path) + if err != nil { + return nil, fmt.Errorf("caller-tool task secret: %w", err) + } + if info.Mode().Perm()&0o077 != 0 { + return nil, fmt.Errorf("caller-tool task secret %s must not be accessible by group or other users", path) + } + payload, err := os.ReadFile(path) + if err != nil { + return nil, err + } + if err := os.Remove(path); err != nil { + return nil, fmt.Errorf("consume caller-tool task secret: %w", err) + } + var secret callerToolEndpointSecret + if err := json.Unmarshal(payload, &secret); err != nil { + return nil, fmt.Errorf("decode caller-tool task secret: %w", err) + } + endpoint := &api.CallerToolEndpoint{Name: secret.Name, URL: secret.URL, Headers: secret.Headers} + if err := endpoint.Validate(); err != nil { + return nil, err + } + return endpoint, nil +} + +func callerToolGrantToWire(grant CallerToolGrant) (callerToolGrantWire, error) { + if err := ValidateTaskID(grant.Task); err != nil { + return callerToolGrantWire{}, err + } + parsed, err := url.Parse(grant.Endpoint.URL) + if err != nil { + return callerToolGrantWire{}, err + } + credential, ok := strings.CutPrefix(grant.Endpoint.Headers["Authorization"], "Bearer ") + if !ok || strings.TrimSpace(credential) == "" { + return callerToolGrantWire{}, fmt.Errorf("delegated caller-tool endpoint has no bearer credential") + } + if grant.Endpoint.Headers[callertools.TaskHeader] != grant.Task || + grant.Endpoint.Headers[callertools.AgentHeader] != grant.Agent { + return callerToolGrantWire{}, fmt.Errorf("delegated caller-tool endpoint binding does not match its task and agent") + } + return callerToolGrantWire{ + Task: grant.Task, Agent: grant.Agent, Name: grant.Endpoint.Name, + Route: parsed.EscapedPath(), Credential: strings.TrimSpace(credential), ExpiresAt: grant.ExpiresAt, + }, nil +} + +func validateCallerToolGrant(grant callerToolGrantWire) error { + if err := ValidateTaskID(grant.Task); err != nil { + return err + } + if strings.TrimSpace(grant.Agent) == "" || strings.TrimSpace(grant.Name) == "" || strings.TrimSpace(grant.Credential) == "" { + return fmt.Errorf("caller-tool grant requires agent, endpoint name, and credential") + } + trimmed, ok := strings.CutPrefix(grant.Route, callertools.RemoteEndpointPrefix) + parts := strings.Split(trimmed, "/") + if !ok || len(parts) != 2 || parts[0] == "" || parts[1] != "mcp" { + return fmt.Errorf("caller-tool grant route is not a Captain remote endpoint") + } + if !grant.ExpiresAt.After(time.Now()) { + return fmt.Errorf("caller-tool grant has expired") + } + return nil +} + +func newCallerToolControlRequest( + ctx context.Context, + method string, + target TransportTarget, + suffix string, + body io.Reader, +) (*http.Request, error) { + if EndpointScheme(target.URL) != "https" { + return nil, fmt.Errorf("delegated caller tools require an HTTPS git-agent endpoint, got %s", target.URL) + } + if target.Token.IsEmpty() { + return nil, fmt.Errorf("delegated caller tools require the sidecar's authenticated dispatch channel") + } + parsed, err := url.Parse(target.URL) + if err != nil { + return nil, err + } + parsed.Path = CallerToolPath + suffix + parsed.RawPath, parsed.RawQuery, parsed.Fragment = "", "", "" + request, err := http.NewRequestWithContext(ctx, method, parsed.String(), body) + if err != nil { + return nil, err + } + request.Header.Set("Authorization", "Bearer "+target.Token.Value()) + return request, nil +} + +func doCallerToolControl(client *http.Client, request *http.Request) error { + response, err := client.Do(request) + if err != nil { + return fmt.Errorf("caller-tool sidecar exchange: %w", err) + } + defer response.Body.Close() + payload, _ := io.ReadAll(io.LimitReader(response.Body, maxCallerToolGrantBytes)) + if response.StatusCode < 200 || response.StatusCode >= 300 { + return fmt.Errorf("caller-tool sidecar exchange returned %s: %s", response.Status, strings.TrimSpace(string(payload))) + } + return nil +} + +func writeCallerToolSecret(root, task string, endpoint callerToolEndpointSecret) error { + payload, err := json.Marshal(endpoint) + if err != nil { + return err + } + dir := taskStateDir(root, task) + if err := os.MkdirAll(dir, 0o700); err != nil { + return err + } + return writeFileAtomic(filepath.Join(dir, callerToolSecretFile), payload, 0o600) +} + +func removeCallerToolSecret(root, task string) error { + err := os.Remove(callerToolSecretPath(root, task)) + if os.IsNotExist(err) { + return nil + } + return err +} + +func callerToolSecretPath(root, task string) string { + return filepath.Join(taskStateDir(root, task), callerToolSecretFile) +} + +// removeStaleCallerToolSecrets fails closed after a sidecar restart: in-memory +// proxy sessions cannot be recovered, so their unconsumed endpoint files must +// not outlive the process that authenticated them. +func removeStaleCallerToolSecrets(root string) (int, error) { + tasks := filepath.Join(root, "captain", "tasks") + entries, err := os.ReadDir(tasks) + if os.IsNotExist(err) { + return 0, nil + } + if err != nil { + return 0, fmt.Errorf("scan stale caller-tool task secrets: %w", err) + } + removed := 0 + for _, entry := range entries { + if !entry.IsDir() || ValidateTaskID(entry.Name()) != nil { + continue + } + err := os.Remove(filepath.Join(tasks, entry.Name(), callerToolSecretFile)) + if os.IsNotExist(err) { + continue + } + if err != nil { + return removed, fmt.Errorf("remove stale caller-tool task secret for %s: %w", entry.Name(), err) + } + removed++ + } + return removed, nil +} + +func parseHTTPSBase(raw, name string) (*url.URL, error) { + parsed, err := url.Parse(strings.TrimSpace(raw)) + if err != nil || parsed.Scheme != "https" || parsed.Host == "" { + return nil, fmt.Errorf("%s %q must use https://", name, raw) + } + return parsed, nil +} + +func parseCallerToolBase(raw string) (*url.URL, error) { + parsed, err := url.Parse(strings.TrimSpace(raw)) + if err != nil || parsed.Host == "" { + return nil, fmt.Errorf("caller-tool endpoint base %q must be an absolute URL", raw) + } + probe := *parsed + probe.Path = CallerToolPath + "/probe/mcp" + probe.RawPath, probe.RawQuery, probe.Fragment = "", "", "" + if err := (api.CallerToolEndpoint{ + Name: "captain", URL: probe.String(), + Headers: map[string]string{"Authorization": "Bearer probe"}, + }).Validate(); err != nil { + return nil, err + } + return parsed, nil +} + +func copyHTTPHeaders(destination, source http.Header) { + connectionHeaders := map[string]bool{} + for _, value := range source.Values("Connection") { + for _, name := range strings.Split(value, ",") { + connectionHeaders[http.CanonicalHeaderKey(strings.TrimSpace(name))] = true + } + } + for name, values := range source { + if isHopByHopHeader(name) || connectionHeaders[http.CanonicalHeaderKey(name)] { + continue + } + for _, value := range values { + destination.Add(name, value) + } + } +} + +func removeHopByHopHeaders(headers http.Header) { + for _, value := range headers.Values("Connection") { + for _, name := range strings.Split(value, ",") { + headers.Del(strings.TrimSpace(name)) + } + } + for name := range headers { + if isHopByHopHeader(name) { + headers.Del(name) + } + } +} + +func isHopByHopHeader(name string) bool { + switch http.CanonicalHeaderKey(name) { + case "Connection", "Keep-Alive", "Proxy-Authenticate", "Proxy-Authorization", "Te", "Trailer", "Transfer-Encoding", "Upgrade": + return true + default: + return false + } +} + +func writeCallerToolRejection(w http.ResponseWriter) { + w.Header().Set("WWW-Authenticate", "Bearer") + http.Error(w, "invalid caller-tool credential", http.StatusUnauthorized) +} diff --git a/pkg/gitagent/dispatch.go b/pkg/gitagent/dispatch.go index 6b04b82b..6e85cf1c 100644 --- a/pkg/gitagent/dispatch.go +++ b/pkg/gitagent/dispatch.go @@ -24,6 +24,9 @@ type TaskPayload struct { Prompt string `json:"prompt"` System string `json:"system,omitempty"` Model string `json:"model,omitempty"` + // CallerTools announces that the runner must consume the separately + // delivered task secret. No endpoint or credential enters task.json. + CallerTools bool `json:"callerTools,omitempty"` // Backend and Effort record the runtime the supervisor resolved, so the // agent does not re-resolve the model against its own defaults. Backend string `json:"backend,omitempty"` diff --git a/pkg/gitagent/hookmain.go b/pkg/gitagent/hookmain.go index aa4b5067..bd01f388 100644 --- a/pkg/gitagent/hookmain.go +++ b/pkg/gitagent/hookmain.go @@ -31,6 +31,7 @@ type HookRuntime struct { // test binary and cannot activate in production. HookSandbox string `json:"hookSandbox,omitempty"` AgentCommand string `json:"agentCommand,omitempty"` + Agent string `json:"-"` // sidecar-local enrollment identity RealRepo string `json:"realRepo,omitempty"` // mailbox-local integration target Relay RelayTarget `json:"relay,omitempty"` // sidecar: supervisor base endpoint } diff --git a/pkg/gitagent/httpclient.go b/pkg/gitagent/httpclient.go index 24ab749c..56dc2a95 100644 --- a/pkg/gitagent/httpclient.go +++ b/pkg/gitagent/httpclient.go @@ -5,13 +5,53 @@ package gitagent import ( + "crypto/tls" + "crypto/x509" "fmt" + "net/http" "net/url" + "os" "strings" "github.com/flanksource/clicky/text" ) +// HTTPSClient builds a Go HTTP client with the same optional trust anchor and +// public-key pin used by Git-agent relay pushes. +func HTTPSClient(caPath, pinnedPublicKey string) (*http.Client, error) { + tlsConfig := &tls.Config{MinVersion: tls.VersionTLS12} + if strings.TrimSpace(caPath) != "" { + roots, err := x509.SystemCertPool() + if err != nil || roots == nil { + roots = x509.NewCertPool() + } + pem, err := os.ReadFile(caPath) + if err != nil { + return nil, fmt.Errorf("read HTTPS CA certificate: %w", err) + } + if !roots.AppendCertsFromPEM(pem) { + return nil, fmt.Errorf("HTTPS CA file %s contains no certificates", caPath) + } + tlsConfig.RootCAs = roots + } + if pin := strings.TrimSpace(pinnedPublicKey); pin != "" { + tlsConfig.VerifyConnection = func(state tls.ConnectionState) error { + if len(state.PeerCertificates) == 0 { + return fmt.Errorf("HTTPS endpoint presented no certificate") + } + actual, err := publicKeyPin(state.PeerCertificates[0]) + if err != nil { + return err + } + if actual != pin { + return fmt.Errorf("HTTPS endpoint public key %s does not match pinned %s", actual, pin) + } + return nil + } + } + return &http.Client{Transport: &http.Transport{TLSClientConfig: tlsConfig}}, nil +} + // TransportTarget is everything a push needs to reach an endpoint. The URL's // scheme selects the transport; the other transport's fields are ignored. type TransportTarget struct { diff --git a/pkg/sandbox/adapter/gitagent.go b/pkg/sandbox/adapter/gitagent.go index 1bc3364b..42bf898e 100644 --- a/pkg/sandbox/adapter/gitagent.go +++ b/pkg/sandbox/adapter/gitagent.go @@ -63,6 +63,59 @@ func (g *gitAgentSandbox) Execute(ctx context.Context, spec api.Spec) (*api.Resp if err != nil { return nil, err } + task := "" + var delegation *api.CallerToolDelegation + callerToolsDelivered := false + if len(g.cfg.CallerTools) > 0 { + if g.cfg.CallerToolEndpoint == nil { + return nil, fmt.Errorf("delegated caller tools require a supervisor caller-tool endpoint") + } + if !api.SupportsCallerTools(spec.Backend) { + return nil, fmt.Errorf("remote backend %q does not support delegated caller tools", spec.Backend) + } + if spec.Permissions.MCP.Disabled { + return nil, fmt.Errorf("delegated caller tools require MCP but MCP is disabled") + } + if gitagent.EndpointScheme(target.url) != "https" { + return nil, fmt.Errorf("delegated caller tools require an HTTPS git-agent sidecar; agent %q uses %s", target.agent, target.url) + } + if err := g.cfg.CallerToolEndpoint.Validate(); err != nil { + return nil, fmt.Errorf("delegate caller tools: %w", err) + } + if g.cfg.CallerToolEndpoint.Delegate == nil { + return nil, fmt.Errorf("caller-tool endpoint does not support task-scoped remote delegation") + } + task, err = gitagent.NewTaskID() + if err != nil { + return nil, err + } + expiresAt := time.Now().Add(target.waitTimeout) + if deadline, ok := ctx.Deadline(); ok && deadline.Before(expiresAt) { + expiresAt = deadline + } + delegation, err = g.cfg.CallerToolEndpoint.Delegate(ctx, api.CallerToolBinding{ + TaskID: task, Agent: target.agent, ExpiresAt: expiresAt, + ToolNames: append([]string(nil), g.cfg.CallerTools...), + }) + if err != nil { + return nil, fmt.Errorf("issue delegated caller tools: %w", err) + } + if delegation == nil || delegation.Revoke == nil { + return nil, fmt.Errorf("issue delegated caller tools: endpoint returned an incomplete delegation") + } + defer delegation.Revoke() + defer func() { + cleanupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = gitagent.RevokeCallerTools(cleanupCtx, target.transport(), task) + }() + if err := gitagent.RegisterCallerTools(ctx, target.transport(), gitagent.CallerToolGrant{ + Task: task, Agent: target.agent, Endpoint: delegation.Endpoint, ExpiresAt: expiresAt, + }); err != nil { + return nil, fmt.Errorf("deliver delegated caller tools: %w", err) + } + callerToolsDelivered = true + } repoDir := spec.Cwd() mailbox, err := gitagent.EnsureMailbox(ctx, target.mailboxRoot, repoDir) if err != nil { @@ -104,12 +157,14 @@ func (g *gitAgentSandbox) Execute(ctx context.Context, spec api.Spec) (*api.Resp KeyPath: target.keyPath, Relay: target.relay, Policy: target.policy, + Task: task, // The resolved backend travels with the model: the agent must run the // runtime the supervisor selected, not re-resolve the name against its // own defaults and quietly pick a different one. TaskPayload: gitagent.TaskPayload{ Prompt: prompt, System: system, Model: spec.Name, Backend: string(spec.Backend), Effort: spec.Effort, Timeout: timeout, + CallerTools: callerToolsDelivered, }, HooksJSON: hooksJSON, }) @@ -175,6 +230,13 @@ type gitAgentTarget struct { waitTimeout time.Duration } +func (target gitAgentTarget) transport() gitagent.TransportTarget { + return gitagent.TransportTarget{ + URL: target.url, KeyPath: target.keyPath, HostFingerprint: target.hostFingerprint, + Token: target.token, + } +} + // resolveTarget picks the enrolled agent — pinned by the spec's sandbox.agent, // or the sole enrolled one — and assembles transport details from options. func (g *gitAgentSandbox) resolveTarget() (*gitAgentTarget, error) { From d5ce90f75d7e04562122c6479e888802535ddaa6 Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Fri, 28 Aug 2026 18:19:46 +0000 Subject: [PATCH 2/6] feat(chat): configure remote git-agent delegation Chat requests previously had no supported way to select a remote Git agent and delegated caller tools, leaving authenticated delegation reachable only through lower-level runtime configuration. Expose sandbox, agent, and tool selection in chat; resolve that untrusted selection server-side; preserve ordered tool policy through execution and approval continuations; and validate caller-tool input before approval brokerage. Consume clicky-ui 0.3.27 so overlapping Agent and CLI model selections remain coherent. Amp-Thread-ID: https://ampcode.com/threads/T-01a03e3c-39ce-714c-b7fe-60010c4f452f --- pkg/ai/callertools/runtime.go | 5 +- pkg/aichat/approval_execution.go | 5 + pkg/aichat/messages.go | 2 + pkg/aichat/provider_config.go | 15 ++ pkg/aichat/service.go | 26 +- pkg/aichat/wire.go | 2 + pkg/cli/serve_chat.go | 7 + pkg/cli/webapp/package.json | 2 +- pkg/cli/webapp/pnpm-lock.yaml | 10 +- pkg/cli/webapp/src/ChatLayer.tsx | 169 ++++++++++--- pkg/cli/webapp/src/RemoteAgentDelegation.tsx | 243 +++++++++++++++++++ 11 files changed, 443 insertions(+), 43 deletions(-) create mode 100644 pkg/cli/webapp/src/RemoteAgentDelegation.tsx diff --git a/pkg/ai/callertools/runtime.go b/pkg/ai/callertools/runtime.go index 228b3480..c1ea599c 100644 --- a/pkg/ai/callertools/runtime.go +++ b/pkg/ai/callertools/runtime.go @@ -170,7 +170,6 @@ func New(options Options) (*Runtime, error) { "1.0.0", server.WithToolCapabilities(false), server.WithToolFilter(runtime.filterTools), - server.WithInputSchemaValidation(), ) for _, definition := range definitions { runtime.definitions[definition.Name] = definition @@ -383,6 +382,10 @@ func (r *Runtime) handler(definition api.ToolDefinition) server.ToolHandlerFunc r.auditCall(ctx, definition.Name, "denied", "invalid_tool_use_id") return mcp.NewToolResultError(err.Error()), nil } + if err := r.validateInput(definition.Name, input); err != nil { + r.auditCall(ctx, definition.Name, "denied", "invalid_input") + return mcp.NewToolResultError(err.Error()), nil + } if definition.NeedsApproval() { if r.canUseTool == nil { r.auditCall(ctx, definition.Name, "denied", "approval_broker_unavailable") diff --git a/pkg/aichat/approval_execution.go b/pkg/aichat/approval_execution.go index 6509a6de..e9271bf0 100644 --- a/pkg/aichat/approval_execution.go +++ b/pkg/aichat/approval_execution.go @@ -77,6 +77,11 @@ func (s *Service) resumeToolApproval(ctx context.Context, threadID string, conti config.SessionID = continuation.Spec.SessionID config.CaptainSessionID = execution.CaptainSessionID() config.Tools = definitions + config.CallerTools = execution.CallerTools() + config, err = s.applySandboxSelection(ctx, config, continuation.Spec.Sandbox) + if err != nil { + return false, err + } config, err = s.prepareProviderConfig(ctx, config) if err != nil { return false, err diff --git a/pkg/aichat/messages.go b/pkg/aichat/messages.go index fba0bf89..53134283 100644 --- a/pkg/aichat/messages.go +++ b/pkg/aichat/messages.go @@ -78,8 +78,10 @@ func requestSpec(request ChatRequest, profile RuntimeProfile, attachments map[pa Model: override, Budget: request.Budget, ToolPreferences: request.ToolPreferences, + ToolPolicy: request.ToolPolicy, ToolApproval: request.ToolApproval, Permissions: api.Permissions{Mode: request.PermissionMode}, + Sandbox: request.Sandbox, SessionID: request.ProviderSessionID, }} layers := append([]api.SpecLayer(nil), profile.Resolved.Trace...) diff --git a/pkg/aichat/provider_config.go b/pkg/aichat/provider_config.go index 219daed5..cedd137c 100644 --- a/pkg/aichat/provider_config.go +++ b/pkg/aichat/provider_config.go @@ -196,3 +196,18 @@ func (s *Service) prepareProviderConfig(ctx context.Context, config api.Config) } return config, nil } + +func (s *Service) applySandboxSelection(ctx context.Context, config api.Config, ref *api.SandboxRef) (api.Config, error) { + if ref == nil { + return config, nil + } + if s.options.ResolveSandbox == nil { + return api.Config{}, fmt.Errorf("chat sandbox selection is not supported by this host") + } + selection, err := s.options.ResolveSandbox(ctx, *ref) + if err != nil { + return api.Config{}, fmt.Errorf("resolve chat sandbox %q: %w", ref.Backend, err) + } + config.SandboxSelection = selection + return config, nil +} diff --git a/pkg/aichat/service.go b/pkg/aichat/service.go index 99bb0101..4492c0f4 100644 --- a/pkg/aichat/service.go +++ b/pkg/aichat/service.go @@ -69,6 +69,10 @@ type ServiceOptions struct { Attachments AttachmentResolver Threads ThreadStoreProvider Authority ExecutionAuthority + // ResolveSandbox turns an untrusted serialized selector into the runtime + // adapter configuration owned by the host. Nil rejects request-level sandbox + // selection rather than silently running without one. + ResolveSandbox func(context.Context, api.SandboxRef) (*api.SandboxConfig, error) // ToolStrategies is how this deployment reads a tool's own facts when no rule // mentions it, weakest first. Nil takes api.DefaultStrategies (HTTP method, // then safety hints). It is one chain for every tool source rather than one @@ -102,10 +106,22 @@ func (s *Service) Handler() http.Handler { mux.HandleFunc("POST /api/chat", s.handleChat) mux.HandleFunc("GET /api/chat/models", s.handleModels) mux.HandleFunc("GET /api/chat/runtimes", s.handleRuntimes) + mux.HandleFunc("GET /api/chat/tools", s.handleTools) s.registerThreadRoutes(mux) return mux } +func (s *Service) handleTools(w http.ResponseWriter, request *http.Request) { + set, err := s.loadTools(request.Context()) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if err := writeJSON(w, http.StatusOK, ToolCatalogResponse{Tools: set.Catalog}); err != nil { + serviceLog.Errorf("write chat tools response: %v", err) + } +} + func (s *Service) handleRuntimes(w http.ResponseWriter, request *http.Request) { profile, err := s.runtimeProfile(request.Context()) if err != nil { @@ -231,12 +247,10 @@ func (s *Service) handleChat(w http.ResponseWriter, request *http.Request) { config.SessionID = spec.SessionID config.CaptainSessionID = chat.ThreadID config.Tools = definitions - if config.SandboxSelection != nil && spec.Sandbox != nil { - selection := *config.SandboxSelection - selection.Agent = spec.Sandbox.Agent - selection.CallerTools = append([]string(nil), spec.Sandbox.CallerTools...) - selection.Policy = spec.Sandbox.Policy - config.SandboxSelection = &selection + config, err = s.applySandboxSelection(request.Context(), config, spec.Sandbox) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return } config, err = s.prepareProviderConfig(request.Context(), config) if err != nil { diff --git a/pkg/aichat/wire.go b/pkg/aichat/wire.go index 9cd63e59..ae2153f1 100644 --- a/pkg/aichat/wire.go +++ b/pkg/aichat/wire.go @@ -23,7 +23,9 @@ type ChatRequest struct { Temperature *float64 `json:"temperature,omitempty"` Budget api.Budget `json:"budget,omitempty"` ToolPreferences api.ToolPreferences `json:"toolPreferences,omitempty"` + ToolPolicy api.PermissionPolicy `json:"toolPolicy,omitempty"` PermissionMode api.PermissionMode `json:"permissionMode,omitempty"` + Sandbox *api.SandboxRef `json:"sandbox,omitempty"` ToolApproval *api.ToolApprovalResume `json:"-"` Context string `json:"context,omitempty"` diff --git a/pkg/cli/serve_chat.go b/pkg/cli/serve_chat.go index 5f22b7f1..a8efb3ab 100644 --- a/pkg/cli/serve_chat.go +++ b/pkg/cli/serve_chat.go @@ -69,6 +69,13 @@ func newCaptainChatService( Threads: aichat.ThreadStoreProviderFunc(contextThreadStore), Authority: authority, Attachments: chatAttachmentResolver{store: attachmentStore}, + ResolveSandbox: func(_ context.Context, ref api.SandboxRef) (*api.SandboxConfig, error) { + selection, err := resolveSandboxSelection("", &ref, loadSavedConfig().Sandbox) + if err != nil { + return nil, err + } + return sandboxSelectionConfig(selection, &ref), nil + }, }) return chat, mcpTools, nil } diff --git a/pkg/cli/webapp/package.json b/pkg/cli/webapp/package.json index 76875dce..896a19b8 100644 --- a/pkg/cli/webapp/package.json +++ b/pkg/cli/webapp/package.json @@ -11,7 +11,7 @@ }, "dependencies": { "@ai-sdk/react": "^3.0.201", - "@flanksource/clicky-ui": "0.3.25", + "@flanksource/clicky-ui": "0.3.27", "@shikijs/langs": "^1.24.0", "@shikijs/themes": "^1.24.0", "@shikijs/transformers": "^1.24.0", diff --git a/pkg/cli/webapp/pnpm-lock.yaml b/pkg/cli/webapp/pnpm-lock.yaml index 41d2ddfa..8b8bf01f 100644 --- a/pkg/cli/webapp/pnpm-lock.yaml +++ b/pkg/cli/webapp/pnpm-lock.yaml @@ -16,8 +16,8 @@ importers: specifier: ^3.0.201 version: 3.0.216(react@18.3.1)(zod@4.4.3) '@flanksource/clicky-ui': - specifier: 0.3.25 - version: 0.3.25(@ai-sdk/react@3.0.216(react@18.3.1)(zod@4.4.3))(@babel/core@7.29.7)(@babel/template@7.29.7)(@shikijs/langs@1.29.2)(@shikijs/themes@1.29.2)(@shikijs/transformers@1.29.2)(@types/react@19.2.17)(ai@6.0.214(zod@4.4.3))(marked@15.0.12)(monaco-editor@0.48.0)(react-dom@18.3.1(react@18.3.1))(react-rnd@10.5.3(react-dom@18.3.1(react@18.3.1))(react@18.3.1))(react@18.3.1)(recharts@3.10.1(@types/react@19.2.17)(react-dom@18.3.1(react@18.3.1))(react-is@17.0.2)(react@18.3.1)(redux@5.0.1))(shiki@1.29.2)(streamdown@2.5.0(react-dom@18.3.1(react@18.3.1))(react@18.3.1))(tailwindcss@4.3.1) + specifier: 0.3.27 + version: 0.3.27(@ai-sdk/react@3.0.216(react@18.3.1)(zod@4.4.3))(@babel/core@7.29.7)(@babel/template@7.29.7)(@shikijs/langs@1.29.2)(@shikijs/themes@1.29.2)(@shikijs/transformers@1.29.2)(@types/react@19.2.17)(ai@6.0.214(zod@4.4.3))(marked@15.0.12)(monaco-editor@0.48.0)(react-dom@18.3.1(react@18.3.1))(react-rnd@10.5.3(react-dom@18.3.1(react@18.3.1))(react@18.3.1))(react@18.3.1)(recharts@3.10.1(@types/react@19.2.17)(react-dom@18.3.1(react@18.3.1))(react-is@17.0.2)(react@18.3.1)(redux@5.0.1))(shiki@1.29.2)(streamdown@2.5.0(react-dom@18.3.1(react@18.3.1))(react@18.3.1))(tailwindcss@4.3.1) '@shikijs/langs': specifier: ^1.24.0 version: 1.29.2 @@ -426,8 +426,8 @@ packages: cpu: [x64] os: [win32] - '@flanksource/clicky-ui@0.3.25': - resolution: {integrity: sha512-YsPyxwJxpYDn4GFcWKSDrA1VDjsR0REtnEvNyUzTfsfQ5piOJnuG4FJuDwh3z2YhHssfl2uS/BKoetG5aSEeAw==} + '@flanksource/clicky-ui@0.3.27': + resolution: {integrity: sha512-+6qYAGRDxU4Ru6NkwqIVRIh8gr+gHg/r7+JFyBozIBsOzjsXhiZuytIcpD2NS/yZcnr5OK/KlLygM/57izKEFA==} peerDependencies: '@ai-sdk/react': ^3.0.0 '@mdxeditor/editor': ^4.0.4 @@ -2771,7 +2771,7 @@ snapshots: '@esbuild/win32-x64@0.28.1': optional: true - '@flanksource/clicky-ui@0.3.25(@ai-sdk/react@3.0.216(react@18.3.1)(zod@4.4.3))(@babel/core@7.29.7)(@babel/template@7.29.7)(@shikijs/langs@1.29.2)(@shikijs/themes@1.29.2)(@shikijs/transformers@1.29.2)(@types/react@19.2.17)(ai@6.0.214(zod@4.4.3))(marked@15.0.12)(monaco-editor@0.48.0)(react-dom@18.3.1(react@18.3.1))(react-rnd@10.5.3(react-dom@18.3.1(react@18.3.1))(react@18.3.1))(react@18.3.1)(recharts@3.10.1(@types/react@19.2.17)(react-dom@18.3.1(react@18.3.1))(react-is@17.0.2)(react@18.3.1)(redux@5.0.1))(shiki@1.29.2)(streamdown@2.5.0(react-dom@18.3.1(react@18.3.1))(react@18.3.1))(tailwindcss@4.3.1)': + '@flanksource/clicky-ui@0.3.27(@ai-sdk/react@3.0.216(react@18.3.1)(zod@4.4.3))(@babel/core@7.29.7)(@babel/template@7.29.7)(@shikijs/langs@1.29.2)(@shikijs/themes@1.29.2)(@shikijs/transformers@1.29.2)(@types/react@19.2.17)(ai@6.0.214(zod@4.4.3))(marked@15.0.12)(monaco-editor@0.48.0)(react-dom@18.3.1(react@18.3.1))(react-rnd@10.5.3(react-dom@18.3.1(react@18.3.1))(react@18.3.1))(react@18.3.1)(recharts@3.10.1(@types/react@19.2.17)(react-dom@18.3.1(react@18.3.1))(react-is@17.0.2)(react@18.3.1)(redux@5.0.1))(shiki@1.29.2)(streamdown@2.5.0(react-dom@18.3.1(react@18.3.1))(react@18.3.1))(tailwindcss@4.3.1)': dependencies: '@codemirror/autocomplete': 6.20.3 '@codemirror/commands': 6.11.0 diff --git a/pkg/cli/webapp/src/ChatLayer.tsx b/pkg/cli/webapp/src/ChatLayer.tsx index d7a52383..26993cd3 100644 --- a/pkg/cli/webapp/src/ChatLayer.tsx +++ b/pkg/cli/webapp/src/ChatLayer.tsx @@ -1,47 +1,156 @@ -import { useMemo } from "react"; -import { ChatFab, ChatWindowLayer } from "@flanksource/clicky-ui/ai"; -import { clickyOperationsToTools } from "@flanksource/clicky-ui/chat"; -import { useOperations } from "@flanksource/clicky-ui/rpc"; -import { apiClient } from "./api"; +import { useEffect, useRef, useState } from "react"; +import { useQuery } from "@tanstack/react-query"; +import { + ChatFab, + ChatWindow, + useChatWindowManager, + type ChatWindowState, + type SpecRuntimeSandboxCatalog, + type ToolMeta, +} from "@flanksource/clicky-ui/ai"; +import type { ChatModelRuntime } from "@flanksource/clicky-ui/chat"; import { isReadOnlyDbContext } from "./dbContext"; -import { isChatToolOperation } from "./session"; +import { fetchSandboxCatalog } from "./sandboxData"; +import { + fetchChatTools, + RemoteAgentDelegation, + type RemoteAgentSelection, +} from "./RemoteAgentDelegation"; + +const LOCAL_EXECUTION: RemoteAgentSelection = { + backend: "", + agent: "", + callerTools: [], +}; export function ChatLayer() { - const { operations } = useOperations(apiClient); // Every chat action creates or appends to a thread, and a read-only database // context rejects those writes. The composer is withheld entirely rather than // left to fail: the chat transport is built inside clicky-ui, so there is no // per-control disable to reach from here. const readOnly = isReadOnlyDbContext(); - const tools = useMemo( - () => clickyOperationsToTools(operations.filter(isChatToolOperation)), - [operations], - ); + const { panels } = useChatWindowManager(); + const tools = useQuery({ + queryKey: ["chat-tool-catalog"], + queryFn: fetchChatTools, + staleTime: 30_000, + }); + const sandboxes = useQuery({ + queryKey: ["sandbox-catalog"], + queryFn: fetchSandboxCatalog, + staleTime: 30_000, + }); if (readOnly) return null; return ( <> - + {panels.map((panel) => ( + + ))} ); } + +type DelegatingChatWindowProps = { + panel: ChatWindowState; + tools: ToolMeta[]; + toolsLoading: boolean; + toolsError?: string; + sandboxCatalog?: SpecRuntimeSandboxCatalog; + sandboxLoading: boolean; + sandboxError?: string; +}; + +function DelegatingChatWindow({ + panel, + tools, + toolsLoading, + toolsError, + sandboxCatalog, + sandboxLoading, + sandboxError, +}: DelegatingChatWindowProps) { + const [selection, setSelection] = useState(LOCAL_EXECUTION); + const [runtimeMode, setRuntimeMode] = useState(); + const previousThread = useRef(panel.threadId); + + useEffect(() => { + if (previousThread.current && previousThread.current !== panel.threadId) { + setSelection(LOCAL_EXECUTION); + } + previousThread.current = panel.threadId; + }, [panel.threadId]); + + const sandbox = selection.backend + ? { + backend: selection.backend, + ...(selection.agent ? { agent: selection.agent } : {}), + ...(selection.callerTools.length > 0 + ? { callerTools: selection.callerTools } + : {}), + } + : undefined; + + return ( + + )} + chat={{ + api: "/api/chat", + modelsApi: "/api/chat/models", + body: sandbox ? { sandbox } : {}, + onRuntimeChange: (runtime) => setRuntimeMode(modeForRuntime(runtime)), + // No defaultModel: the served menu marks captain's own default, so a + // literal here would only go stale or name a disabled model. + enableAttachments: true, + suggestions: [ + "Summarize this run", + "Show recent changed files", + "Check the current repo status", + ], + placeholder: sandbox + ? `Send a task to ${selection.agent || selection.backend}...` + : "Continue the agent session...", + }} + /> + ); +} + +function queryError(error: Error | null): string | undefined { + return error?.message; +} + +function modeForRuntime(runtime: ChatModelRuntime): string | undefined { + if (runtime.backend?.endsWith("-agent")) return "agent"; + if (runtime.backend?.endsWith("-cli")) return "cli"; + if (runtime.backend?.endsWith("-cmux")) return "cmux"; + return runtime.mode; +} diff --git a/pkg/cli/webapp/src/RemoteAgentDelegation.tsx b/pkg/cli/webapp/src/RemoteAgentDelegation.tsx new file mode 100644 index 00000000..433e730c --- /dev/null +++ b/pkg/cli/webapp/src/RemoteAgentDelegation.tsx @@ -0,0 +1,243 @@ +import { useId, useMemo } from "react"; +import { + Button, + Combobox, + DropdownMenu, + type ComboboxOption, +} from "@flanksource/clicky-ui/components"; +import { + Icon, + UiRobotAi, +} from "@flanksource/clicky-ui/data"; +import type { + SpecRuntimeSandboxCatalog, + ToolMeta, +} from "@flanksource/clicky-ui/ai"; +import { normalizeToolPolicy } from "@flanksource/clicky-ui/chat"; + +// This control owns only the serialized delegation request. The supervisor +// resolves policy and issues the task capability after the user submits a turn; +// no credential or durable authority is present in browser state. + +export type RemoteAgentSelection = { + backend: string; + agent: string; + callerTools: string[]; +}; + +type RemoteAgentDelegationProps = { + value: RemoteAgentSelection; + onChange: (value: RemoteAgentSelection) => void; + catalog?: SpecRuntimeSandboxCatalog; + tools: ToolMeta[]; + loadingCatalog: boolean; + loadingTools: boolean; + catalogError?: string; + toolsError?: string; + runtimeMode?: string; +}; + +type GitAgentBackend = { + name: string; + agents: string[]; +}; + +/** Selects the remote Git agent and the exact supervisor tools requested for delegation. */ +export function RemoteAgentDelegation({ + value, + onChange, + catalog, + tools, + loadingCatalog, + loadingTools, + catalogError, + toolsError, + runtimeMode, +}: RemoteAgentDelegationProps) { + const runtimeErrorId = useId(); + const backends = useMemo(() => gitAgentBackends(catalog), [catalog]); + const selected = backends.find((backend) => backend.name === value.backend); + const backendOptions = useMemo( + () => [ + { + value: "", + label: "Run on this supervisor", + description: "Do not dispatch this turn to a remote Git agent.", + }, + ...backends.map((backend) => ({ + value: backend.name, + label: backend.name, + description: backend.agents.length > 0 + ? `${backend.agents.length} dispatchable agent${backend.agents.length === 1 ? "" : "s"}` + : "No dispatchable agents", + disabled: backend.agents.length === 0, + })), + ], + [backends], + ); + const agentOptions = useMemo( + () => (selected?.agents ?? []).map((agent) => ({ value: agent, label: agent })), + [selected], + ); + const toolOptions = useMemo( + () => tools.map((tool) => ({ + value: tool.name, + label: tool.parent ? `${tool.parent}: ${tool.label}` : tool.label, + description: tool.description, + title: tool.name, + })), + [tools], + ); + const remote = value.backend !== ""; + const incompatibleRuntime = runtimeMode !== undefined && runtimeMode !== "agent"; + const triggerTitle = remote + ? `Remote Git agent: ${value.agent || value.backend}; ${value.callerTools.length} delegated tool${value.callerTools.length === 1 ? "" : "s"}` + : "Run on this supervisor"; + + return ( + + + {remote && value.callerTools.length > 0 && ( + + {value.callerTools.length} + + )} + + )} + > + {() => ( +
event.stopPropagation()}> +
+
Remote Git agent
+

+ Dispatch this turn remotely and delegate only the selected supervisor tools. +

+
+ + + + {catalogError && ( +

{catalogError}

+ )} + + {remote && ( + <> + + + + + {incompatibleRuntime && ( + + )} + + {toolsError && ( +

{toolsError}

+ )} + +
+ {value.callerTools.length === 0 + ? "No supervisor tools will be exposed to the remote model." + : `${value.callerTools.length} tool${value.callerTools.length === 1 ? "" : "s"} requested. The supervisor’s effective allow/deny/ask policy can only narrow this list.`} +
+ Captain issues a fresh task-scoped capability at dispatch and never stores it in Git. +
+
+ Caller tools require an Agent runtime; CLI and cmux runs cannot use them. +
+
+ + )} +
+ )} + + ); +} + +/** Loads the canonical tools backed by the supervisor's live handlers. */ +export async function fetchChatTools(): Promise { + const response = await fetch("/api/chat/tools", { + headers: { Accept: "application/json" }, + }); + if (!response.ok) { + const message = (await response.text()).trim(); + throw new Error(message || `Tool catalog failed with ${response.status}`); + } + const body = (await response.json()) as { tools?: ChatToolCatalogEntry[] }; + return (body.tools ?? []).map((tool) => ({ + ...tool, + label: tool.title || tool.operationName || tool.name, + defaultPermission: normalizeToolPolicy(tool.defaultPermission), + })); +} + +type ChatToolCatalogEntry = Omit & { + title?: string; + defaultPermission?: string; +}; + +function gitAgentBackends(catalog?: SpecRuntimeSandboxCatalog): GitAgentBackend[] { + const kind = catalog?.kinds?.find((item) => item.kind === "git-agent"); + if (!kind) return []; + const backends = kind.backends ?? []; + if (backends.length === 0) { + return [{ name: kind.kind, agents: [] }]; + } + return backends.map((backend) => ({ + name: backend.name, + agents: (backend.agents ?? []) + .filter((agent) => agent.dispatchable && (!agent.status || agent.status === "enrolled")) + .map((agent) => agent.name), + })); +} From 06025f26c7ea896bb24d1a2b009b876d93803e2c Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Fri, 28 Aug 2026 18:44:12 +0000 Subject: [PATCH 3/6] fix(git-agent): confine caller-tool secret paths CodeQL traced task-derived values into ordinary filesystem path operations used for transient capability secrets. Task IDs were constrained, but the storage boundary still depended on that validation and could follow a replaced state-directory symlink. Create, read, remove, and clean up those secrets through os.Root with exclusive 0600 files, confining every task-relative name to the sidecar repository. Amp-Thread-ID: https://ampcode.com/threads/T-01a03e3c-39ce-714c-b7fe-60010c4f452f --- pkg/gitagent/callertools.go | 104 ++++++++++++++++++++++++++++++------ 1 file changed, 89 insertions(+), 15 deletions(-) diff --git a/pkg/gitagent/callertools.go b/pkg/gitagent/callertools.go index 9c7e2cf6..71ef2c35 100644 --- a/pkg/gitagent/callertools.go +++ b/pkg/gitagent/callertools.go @@ -349,22 +349,27 @@ func RevokeCallerTools(ctx context.Context, target TransportTarget, task string) // LoadCallerToolEndpoint consumes the sidecar-delivered secret before the // remote model process starts. func LoadCallerToolEndpoint(sidecarRepo, task string) (*api.CallerToolEndpoint, error) { - if err := ValidateTaskID(task); err != nil { + name, err := callerToolSecretName(task) + if err != nil { return nil, err } - path := callerToolSecretPath(sidecarRepo, task) - info, err := os.Stat(path) + root, err := os.OpenRoot(sidecarRepo) + if err != nil { + return nil, fmt.Errorf("open caller-tool secret store: %w", err) + } + defer root.Close() + info, err := root.Stat(name) if err != nil { return nil, fmt.Errorf("caller-tool task secret: %w", err) } if info.Mode().Perm()&0o077 != 0 { - return nil, fmt.Errorf("caller-tool task secret %s must not be accessible by group or other users", path) + return nil, fmt.Errorf("caller-tool task secret must not be accessible by group or other users") } - payload, err := os.ReadFile(path) + payload, err := root.ReadFile(name) if err != nil { return nil, err } - if err := os.Remove(path); err != nil { + if err := root.Remove(name); err != nil { return nil, fmt.Errorf("consume caller-tool task secret: %w", err) } var secret callerToolEndpointSecret @@ -458,48 +463,117 @@ func doCallerToolControl(client *http.Client, request *http.Request) error { return nil } +// writeCallerToolSecret confines the validated task-relative name with +// os.Root, so a sidecar-state symlink cannot redirect capability material. func writeCallerToolSecret(root, task string, endpoint callerToolEndpointSecret) error { + name, err := callerToolSecretName(task) + if err != nil { + return err + } payload, err := json.Marshal(endpoint) if err != nil { return err } - dir := taskStateDir(root, task) - if err := os.MkdirAll(dir, 0o700); err != nil { + store, err := os.OpenRoot(root) + if err != nil { + return err + } + defer store.Close() + if err := store.MkdirAll(filepath.Dir(name), 0o700); err != nil { + return err + } + file, err := store.OpenFile(name, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + if err != nil { + return err + } + created := true + defer func() { + if created { + _ = store.Remove(name) + } + }() + if _, err := file.Write(payload); err != nil { + _ = file.Close() + return err + } + if err := file.Chmod(0o600); err != nil { + _ = file.Close() + return err + } + if err := file.Sync(); err != nil { + _ = file.Close() + return err + } + if err := file.Close(); err != nil { + return err + } + dir, err := store.Open(filepath.Dir(name)) + if err != nil { + return err + } + defer dir.Close() + if err := dir.Sync(); err != nil { return err } - return writeFileAtomic(filepath.Join(dir, callerToolSecretFile), payload, 0o600) + created = false + return nil } func removeCallerToolSecret(root, task string) error { - err := os.Remove(callerToolSecretPath(root, task)) + name, err := callerToolSecretName(task) + if err != nil { + return err + } + store, err := os.OpenRoot(root) + if os.IsNotExist(err) { + return nil + } + if err != nil { + return err + } + defer store.Close() + err = store.Remove(name) if os.IsNotExist(err) { return nil } return err } -func callerToolSecretPath(root, task string) string { - return filepath.Join(taskStateDir(root, task), callerToolSecretFile) +func callerToolSecretName(task string) (string, error) { + if err := ValidateTaskID(task); err != nil { + return "", err + } + return filepath.Join("captain", "tasks", task, callerToolSecretFile), nil } // removeStaleCallerToolSecrets fails closed after a sidecar restart: in-memory // proxy sessions cannot be recovered, so their unconsumed endpoint files must // not outlive the process that authenticated them. func removeStaleCallerToolSecrets(root string) (int, error) { - tasks := filepath.Join(root, "captain", "tasks") - entries, err := os.ReadDir(tasks) + store, err := os.OpenRoot(root) + if err != nil { + return 0, fmt.Errorf("open caller-tool secret store: %w", err) + } + defer store.Close() + tasks := filepath.Join("captain", "tasks") + dir, err := store.Open(tasks) if os.IsNotExist(err) { return 0, nil } if err != nil { return 0, fmt.Errorf("scan stale caller-tool task secrets: %w", err) } + entries, err := dir.ReadDir(-1) + _ = dir.Close() + if err != nil { + return 0, fmt.Errorf("scan stale caller-tool task secrets: %w", err) + } removed := 0 for _, entry := range entries { if !entry.IsDir() || ValidateTaskID(entry.Name()) != nil { continue } - err := os.Remove(filepath.Join(tasks, entry.Name(), callerToolSecretFile)) + err := store.Remove(filepath.Join(tasks, entry.Name(), callerToolSecretFile)) if os.IsNotExist(err) { continue } From 952ffbbf2f5dc4789ae689b9c1e60b2ea27eaac5 Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Fri, 28 Aug 2026 19:02:23 +0000 Subject: [PATCH 4/6] fix(git-agent): address delegation review findings Review found duplicate capability validation, opaque schema-validation placement, a hand-written proxy that could buffer future streamed responses, and imprecise expired-session re-registration. Keep one final liveness check, document the synthetic input and route-carrier invariants, use a streaming ReverseProxy, retire expired sessions without racing replacement grants, and make the runner's caller-tool endpoint argument explicit. Amp-Thread-ID: https://ampcode.com/threads/T-01a03e3c-39ce-714c-b7fe-60010c4f452f --- pkg/ai/callertools/runtime.go | 10 ++-- pkg/cli/gitagent_runtask.go | 6 +- pkg/cli/gitagent_runtask_test.go | 2 +- pkg/gitagent/callertools.go | 99 +++++++++++++------------------- 4 files changed, 46 insertions(+), 71 deletions(-) diff --git a/pkg/ai/callertools/runtime.go b/pkg/ai/callertools/runtime.go index c1ea599c..ccde4b8d 100644 --- a/pkg/ai/callertools/runtime.go +++ b/pkg/ai/callertools/runtime.go @@ -165,6 +165,8 @@ func New(options Options) (*Runtime, error) { cancel: cancel, listener: listener, } + // Validation stays in the handler because out-of-process providers add the + // synthetic tool-use ID that must be removed before strict schema checks. mcpServer := server.NewMCPServer( "captain-caller-tools", "1.0.0", @@ -330,7 +332,9 @@ func (r *Runtime) delegate(_ context.Context, binding api.CallerToolBinding) (*a delegation := &api.CallerToolDelegation{ Endpoint: api.CallerToolEndpoint{ Name: serverName, - URL: "http://127.0.0.1" + RemoteEndpointPrefix + id + endpointPath, + // The sidecar extracts this path and replaces the placeholder origin + // with the enrolled supervisor URL; this value is never dialed. + URL: "http://127.0.0.1" + RemoteEndpointPrefix + id + endpointPath, Headers: map[string]string{ "Authorization": "Bearer " + token, TaskHeader: binding.TaskID, @@ -413,10 +417,6 @@ func (r *Runtime) handler(definition api.ToolDefinition) server.ToolHandlerFunc input = decision.UpdatedInput } } - if err := r.validateActive(callCtx); err != nil { - r.auditCall(ctx, definition.Name, "denied", "capability_inactive") - return mcp.NewToolResultError(err.Error()), nil - } if err := r.validateInput(definition.Name, input); err != nil { r.auditCall(ctx, definition.Name, "denied", "invalid_input") return mcp.NewToolResultError(err.Error()), nil diff --git a/pkg/cli/gitagent_runtask.go b/pkg/cli/gitagent_runtask.go index 9687ebac..fdfa52c7 100644 --- a/pkg/cli/gitagent_runtask.go +++ b/pkg/cli/gitagent_runtask.go @@ -93,12 +93,8 @@ func runTaskPrompt( ctx context.Context, worktree string, payload gitagent.TaskPayload, - callerToolEndpoints ...*api.CallerToolEndpoint, + callerTools *api.CallerToolEndpoint, ) error { - var callerTools *api.CallerToolEndpoint - if len(callerToolEndpoints) > 0 { - callerTools = callerToolEndpoints[0] - } providerOpts := AIProviderOptions{ ModelFlags: aiflags.ModelFlags{Model: payload.Model, Backend: payload.Backend, Effort: string(payload.Effort)}, Sandbox: "none", diff --git a/pkg/cli/gitagent_runtask_test.go b/pkg/cli/gitagent_runtask_test.go index c94b9f00..a8814bfa 100644 --- a/pkg/cli/gitagent_runtask_test.go +++ b/pkg/cli/gitagent_runtask_test.go @@ -24,7 +24,7 @@ func TestRunTaskPromptCarriesSupervisorRuntime(t *testing.T) { Prompt: "make a change", Model: "gpt-5.6-sol", Backend: string(api.BackendCodexCLI), Effort: api.EffortHigh, Timeout: "17m", } - if err := runTaskPrompt(context.Background(), t.TempDir(), payload); err != nil { + if err := runTaskPrompt(context.Background(), t.TempDir(), payload, nil); err != nil { t.Fatal(err) } if captured.Budget.Timeout != "17m" { diff --git a/pkg/gitagent/callertools.go b/pkg/gitagent/callertools.go index 71ef2c35..b6c4997e 100644 --- a/pkg/gitagent/callertools.go +++ b/pkg/gitagent/callertools.go @@ -14,6 +14,7 @@ import ( "fmt" "io" "net/http" + "net/http/httputil" "net/url" "os" "path/filepath" @@ -200,10 +201,22 @@ func (proxy *callerToolProxy) register(w http.ResponseWriter, request *http.Requ tokenHash: sha256.Sum256([]byte(grant.Credential)), expiresAt: grant.ExpiresAt, } proxy.mu.Lock() - if existing := proxy.sessions[grant.Task]; existing != nil && !existing.revoked.Load() { - proxy.mu.Unlock() - http.Error(w, "caller-tool grant already exists for task", http.StatusConflict) - return + var expired *callerToolSession + if existing := proxy.sessions[grant.Task]; existing != nil { + if !existing.revoked.Load() && time.Now().Before(existing.expiresAt) { + proxy.mu.Unlock() + http.Error(w, "caller-tool grant already exists for task", http.StatusConflict) + return + } + delete(proxy.sessions, grant.Task) + if existing.revoked.CompareAndSwap(false, true) { + expired = existing + } + if err := removeCallerToolSecret(proxy.root, grant.Task); err != nil { + proxy.mu.Unlock() + http.Error(w, "remove expired caller-tool task secret", http.StatusInternalServerError) + return + } } if err := writeCallerToolSecret(proxy.root, grant.Task, secret); err != nil { proxy.mu.Unlock() @@ -212,6 +225,9 @@ func (proxy *callerToolProxy) register(w http.ResponseWriter, request *http.Requ } proxy.sessions[grant.Task] = session proxy.mu.Unlock() + if expired != nil { + proxy.log("git-agent caller-tool grant expired task=%s agent=%s", expired.task, expired.agent) + } time.AfterFunc(time.Until(grant.ExpiresAt), func() { proxy.expire(session) }) proxy.log("git-agent caller-tool grant issued task=%s agent=%s expires=%s", grant.Task, grant.Agent, grant.ExpiresAt.UTC().Format(time.RFC3339)) @@ -255,23 +271,21 @@ func (proxy *callerToolProxy) forward(w http.ResponseWriter, request *http.Reque upstreamURL := *proxy.supervisorURL upstreamURL.Path = session.route upstreamURL.RawPath, upstreamURL.RawQuery, upstreamURL.Fragment = "", request.URL.RawQuery, "" - upstream := request.Clone(request.Context()) - upstream.URL = &upstreamURL - upstream.RequestURI = "" - upstream.Host = upstreamURL.Host - removeHopByHopHeaders(upstream.Header) - upstream.Header.Set(callertools.TaskHeader, session.task) - upstream.Header.Set(callertools.AgentHeader, session.agent) - response, err := proxy.client.Do(upstream) - if err != nil { - proxy.log("git-agent caller-tool proxy failed task=%s agent=%s", session.task, session.agent) - http.Error(w, "caller-tool supervisor is unavailable", http.StatusBadGateway) - return + reverseProxy := &httputil.ReverseProxy{ + Rewrite: func(proxyRequest *httputil.ProxyRequest) { + proxyRequest.Out.URL = &upstreamURL + proxyRequest.Out.Host = upstreamURL.Host + proxyRequest.Out.Header.Set(callertools.TaskHeader, session.task) + proxyRequest.Out.Header.Set(callertools.AgentHeader, session.agent) + }, + Transport: proxy.client.Transport, + FlushInterval: -1, + ErrorHandler: func(w http.ResponseWriter, _ *http.Request, _ error) { + proxy.log("git-agent caller-tool proxy failed task=%s agent=%s", session.task, session.agent) + http.Error(w, "caller-tool supervisor is unavailable", http.StatusBadGateway) + }, } - defer response.Body.Close() - copyHTTPHeaders(w.Header(), response.Header) - w.WriteHeader(response.StatusCode) - _, _ = io.Copy(w, response.Body) + reverseProxy.ServeHTTP(w, request) } func (proxy *callerToolProxy) revoke(w http.ResponseWriter, request *http.Request, task string) { @@ -296,13 +310,17 @@ func (proxy *callerToolProxy) revoke(w http.ResponseWriter, request *http.Reques func (proxy *callerToolProxy) expire(session *callerToolSession) { if session.revoked.CompareAndSwap(false, true) { + removed := false proxy.mu.Lock() if proxy.sessions[session.task] == session { delete(proxy.sessions, session.task) + removed = true } proxy.mu.Unlock() proxy.log("git-agent caller-tool grant expired task=%s agent=%s", session.task, session.agent) - _ = removeCallerToolSecret(proxy.root, session.task) + if removed { + _ = removeCallerToolSecret(proxy.root, session.task) + } } } @@ -610,45 +628,6 @@ func parseCallerToolBase(raw string) (*url.URL, error) { return parsed, nil } -func copyHTTPHeaders(destination, source http.Header) { - connectionHeaders := map[string]bool{} - for _, value := range source.Values("Connection") { - for _, name := range strings.Split(value, ",") { - connectionHeaders[http.CanonicalHeaderKey(strings.TrimSpace(name))] = true - } - } - for name, values := range source { - if isHopByHopHeader(name) || connectionHeaders[http.CanonicalHeaderKey(name)] { - continue - } - for _, value := range values { - destination.Add(name, value) - } - } -} - -func removeHopByHopHeaders(headers http.Header) { - for _, value := range headers.Values("Connection") { - for _, name := range strings.Split(value, ",") { - headers.Del(strings.TrimSpace(name)) - } - } - for name := range headers { - if isHopByHopHeader(name) { - headers.Del(name) - } - } -} - -func isHopByHopHeader(name string) bool { - switch http.CanonicalHeaderKey(name) { - case "Connection", "Keep-Alive", "Proxy-Authenticate", "Proxy-Authorization", "Te", "Trailer", "Transfer-Encoding", "Upgrade": - return true - default: - return false - } -} - func writeCallerToolRejection(w http.ResponseWriter) { w.Header().Set("WWW-Authenticate", "Bearer") http.Error(w, "invalid caller-tool credential", http.StatusUnauthorized) From b328a5550aa59dfde05f1eddeae6d0fcdcd9f79c Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Mon, 31 Aug 2026 17:27:23 +0000 Subject: [PATCH 5/6] fix(git-agent): broker delegated caller-tool approvals Relocated agent MCP calls generated tool-use IDs that the supervisor tried to correlate with a local provider stream. Because the provider runs on the remote agent, that event never arrived and approved tools failed after the correlation timeout. Treat authenticated remote MCP calls as the authoritative tool-use observation while preserving durable ask-policy approval. Stream reconstructed use and terminal result events through the supervisor, keep local provider correlation unchanged, and return remote approval failures without hanging. Amp-Thread-ID: https://ampcode.com/threads/T-01a0580d-57bd-769a-8f96-d01d5ba28789 --- pkg/ai/callertools/runtime.go | 122 ++++++++---- pkg/ai/callertools/runtime_ginkgo_test.go | 188 ++++++++++++++++++ pkg/aichat/execution.go | 61 ++++-- pkg/aichat/execution_authority_ginkgo_test.go | 79 ++++++++ pkg/aichat/execution_database.go | 24 ++- .../execution_database_integration_test.go | 175 +++++++++++++++- pkg/api/runtime_config.go | 6 +- pkg/api/runtime_event.go | 3 + 8 files changed, 587 insertions(+), 71 deletions(-) diff --git a/pkg/ai/callertools/runtime.go b/pkg/ai/callertools/runtime.go index ccde4b8d..eda14971 100644 --- a/pkg/ai/callertools/runtime.go +++ b/pkg/ai/callertools/runtime.go @@ -71,6 +71,9 @@ type Options struct { // immediately before a tool handler executes. ValidateCredential func(context.Context) error Audit func(AuditEvent) + // ObserveDelegatedTool publishes the tool-use lifecycle reconstructed from + // an authenticated remote MCP call. Local providers publish their own events. + ObserveDelegatedTool func(context.Context, api.Event) error ApprovalTimeout time.Duration } @@ -78,15 +81,16 @@ type Options struct { // Runtime owns one loopback-only MCP server and its in-memory bearer // credential. Closing it revokes the capability by shutting down the listener. type Runtime struct { - definitions map[string]api.ToolDefinition - schemas map[string]*jsonschema.Schema - canUseTool api.PermissionFunc - validate func(context.Context) error - sessionID string - token string - tokenHash [sha256.Size]byte - expiresAt time.Time - audit func(AuditEvent) + definitions map[string]api.ToolDefinition + schemas map[string]*jsonschema.Schema + canUseTool api.PermissionFunc + validate func(context.Context) error + sessionID string + token string + tokenHash [sha256.Size]byte + expiresAt time.Time + audit func(AuditEvent) + observeDelegatedTool func(context.Context, api.Event) error approvalTimeout time.Duration ctx context.Context @@ -151,19 +155,20 @@ func New(options Options) (*Runtime, error) { } ctx, cancel := context.WithCancel(context.Background()) runtime := &Runtime{ - definitions: make(map[string]api.ToolDefinition, len(definitions)), - schemas: make(map[string]*jsonschema.Schema, len(definitions)), - canUseTool: options.CanUseTool, - validate: options.ValidateCredential, - audit: options.Audit, - sessionID: options.SessionID, - token: token, - tokenHash: sha256.Sum256([]byte(token)), - expiresAt: options.ExpiresAt, - approvalTimeout: options.ApprovalTimeout, - ctx: ctx, - cancel: cancel, - listener: listener, + definitions: make(map[string]api.ToolDefinition, len(definitions)), + schemas: make(map[string]*jsonschema.Schema, len(definitions)), + canUseTool: options.CanUseTool, + validate: options.ValidateCredential, + audit: options.Audit, + observeDelegatedTool: options.ObserveDelegatedTool, + sessionID: options.SessionID, + token: token, + tokenHash: sha256.Sum256([]byte(token)), + expiresAt: options.ExpiresAt, + approvalTimeout: options.ApprovalTimeout, + ctx: ctx, + cancel: cancel, + listener: listener, } // Validation stays in the handler because out-of-process providers add the // synthetic tool-use ID that must be removed before strict schema checks. @@ -373,7 +378,8 @@ func (r *Runtime) handler(definition api.ToolDefinition) server.ToolHandlerFunc stop := context.AfterFunc(r.ctx, cancel) defer stop() defer cancel() - if capability, ok := ctx.Value(delegationContextKey{}).(*delegatedCapability); ok { + capability, delegated := ctx.Value(delegationContextKey{}).(*delegatedCapability) + if delegated { stopCapability := context.AfterFunc(capability.ctx, cancel) defer stopCapability() } @@ -386,60 +392,96 @@ func (r *Runtime) handler(definition api.ToolDefinition) server.ToolHandlerFunc r.auditCall(ctx, definition.Name, "denied", "invalid_tool_use_id") return mcp.NewToolResultError(err.Error()), nil } + delegatedObserved := false + fail := func(message, result, reason string) (*mcp.CallToolResult, error) { + if delegatedObserved { + eventErr := r.observeDelegated(callCtx, api.Event{ + Kind: api.EventToolResult, Tool: definition.Name, ToolCallID: toolUseID, + Text: message, Success: false, Delegated: true, + }) + if eventErr != nil { + r.auditCall(ctx, definition.Name, "error", "event_delivery_failed") + return mcp.NewToolResultErrorf("%s (record delegated tool result: %v)", message, eventErr), nil + } + } + r.auditCall(ctx, definition.Name, result, reason) + return mcp.NewToolResultError(message), nil + } + if delegated { + if err := r.observeDelegated(callCtx, api.Event{ + Kind: api.EventToolUse, Tool: definition.Name, ToolCallID: toolUseID, Input: input, Delegated: true, + }); err != nil { + r.auditCall(ctx, definition.Name, "error", "event_delivery_failed") + return mcp.NewToolResultErrorf("record delegated caller-tool use: %v", err), nil + } + delegatedObserved = true + } if err := r.validateInput(definition.Name, input); err != nil { - r.auditCall(ctx, definition.Name, "denied", "invalid_input") - return mcp.NewToolResultError(err.Error()), nil + return fail(err.Error(), "denied", "invalid_input") } if definition.NeedsApproval() { if r.canUseTool == nil { - r.auditCall(ctx, definition.Name, "denied", "approval_broker_unavailable") - return mcp.NewToolResultError("tool approval is required but no approval broker is configured"), nil + return fail("tool approval is required but no approval broker is configured", "denied", "approval_broker_unavailable") } approvalCtx, approvalCancel := context.WithTimeout(callCtx, r.approvalTimeout) decision, err := r.canUseTool(approvalCtx, api.PermissionRequest{ Tool: definition.Name, Input: input, ToolUseID: toolUseID, - ToolUseIDGenerated: generatedToolUseID, SessionID: r.sessionID, + ToolUseIDGenerated: generatedToolUseID, Delegated: delegated, SessionID: r.sessionID, }) approvalCancel() if err != nil { - r.auditCall(ctx, definition.Name, "denied", "approval_failed") - return mcp.NewToolResultError(err.Error()), nil + return fail(err.Error(), "denied", "approval_failed") } if !decision.Allow { message := decision.Message if message == "" { message = "tool call denied" } - r.auditCall(ctx, definition.Name, "denied", "approval_denied") - return mcp.NewToolResultError(message), nil + return fail(message, "denied", "approval_denied") } if decision.UpdatedInput != nil { input = decision.UpdatedInput } } if err := r.validateInput(definition.Name, input); err != nil { - r.auditCall(ctx, definition.Name, "denied", "invalid_input") - return mcp.NewToolResultError(err.Error()), nil + return fail(err.Error(), "denied", "invalid_input") } if err := r.validateActive(callCtx); err != nil { - r.auditCall(ctx, definition.Name, "denied", "capability_inactive") - return mcp.NewToolResultError(err.Error()), nil + return fail(err.Error(), "denied", "capability_inactive") } output, err := definition.Handler(callCtx, input) if err != nil { - r.auditCall(ctx, definition.Name, "error", "handler_failed") - return mcp.NewToolResultError(err.Error()), nil + return fail(err.Error(), "error", "handler_failed") } result, err := mcp.NewToolResultJSON(output) if err != nil { - r.auditCall(ctx, definition.Name, "error", "result_encoding_failed") - return mcp.NewToolResultErrorf("marshal caller tool %q result: %v", definition.Name, err), nil + return fail(fmt.Sprintf("marshal caller tool %q result: %v", definition.Name, err), "error", "result_encoding_failed") + } + if delegatedObserved { + encoded, err := json.Marshal(output) + if err != nil { + return fail(fmt.Sprintf("marshal caller tool %q result: %v", definition.Name, err), "error", "result_encoding_failed") + } + if err := r.observeDelegated(callCtx, api.Event{ + Kind: api.EventToolResult, Tool: definition.Name, ToolCallID: toolUseID, + Text: string(encoded), Success: true, Delegated: true, + }); err != nil { + r.auditCall(ctx, definition.Name, "error", "event_delivery_failed") + return mcp.NewToolResultErrorf("record delegated caller-tool result: %v", err), nil + } } r.auditCall(ctx, definition.Name, "allowed", "") return result, nil } } +func (r *Runtime) observeDelegated(ctx context.Context, event api.Event) error { + if r.observeDelegatedTool == nil { + return nil + } + return r.observeDelegatedTool(ctx, event) +} + func (r *Runtime) authorizeLocal(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { host, _, err := net.SplitHostPort(request.RemoteAddr) diff --git a/pkg/ai/callertools/runtime_ginkgo_test.go b/pkg/ai/callertools/runtime_ginkgo_test.go index 23cd24f1..eca4941f 100644 --- a/pkg/ai/callertools/runtime_ginkgo_test.go +++ b/pkg/ai/callertools/runtime_ginkgo_test.go @@ -4,6 +4,8 @@ import ( "context" "errors" "net/http" + "net/http/httptest" + "net/url" "sync/atomic" "time" @@ -15,6 +17,7 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + . "github.com/onsi/gomega/gstruct" ) var _ = Describe("Authenticated caller-tool runtime", func() { @@ -78,6 +81,7 @@ var _ = Describe("Authenticated caller-tool runtime", func() { Expect(request.Tool).To(Equal("invoice_update")) Expect(request.SessionID).To(Equal("captain-session-2")) Expect(request.ToolUseID).To(Equal("approval-call-1")) + Expect(request.Delegated).To(BeFalse()) return api.PermissionDecision{Allow: true, UpdatedInput: map[string]any{"status": "approved"}}, nil }, SessionID: "captain-session-2", @@ -308,6 +312,167 @@ var _ = Describe("Authenticated caller-tool runtime", func() { } Expect(values).To(ConsistOf("first", "second")) }) + + It("exposes and executes only the tools selected for a remote task", func(ctx SpecContext) { + remote := httptest.NewServer(callertools.RemoteHandler()) + DeferCleanup(remote.Close) + var hiddenCalls atomic.Int32 + runtime, err := callertools.New(callertools.Options{ + Definitions: []api.ToolDefinition{ + { + Name: "version", DefaultPermission: api.ToolPolicyAllow, + Handler: func(context.Context, map[string]any) (any, error) { + return map[string]any{"version": "test"}, nil + }, + }, + { + Name: "contexts", DefaultPermission: api.ToolPolicyAllow, + Handler: func(context.Context, map[string]any) (any, error) { return []string{"default"}, nil }, + }, + { + Name: "whoami", DefaultPermission: api.ToolPolicyAllow, + Handler: func(context.Context, map[string]any) (any, error) { + hiddenCalls.Add(1) + return "captain", nil + }, + }, + }, + SessionID: "remote-session", + }) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(runtime.Close) + + delegation, err := runtime.Endpoint().Delegate(ctx, api.CallerToolBinding{ + TaskID: "task-1", Agent: "agent-1", ExpiresAt: time.Now().Add(time.Minute), + ToolNames: []string{"version", "contexts"}, + }) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(delegation.Revoke) + client := authenticatedClient(ctx, servedDelegation(remote.URL, delegation.Endpoint)) + DeferCleanup(client.Close) + + tools, err := client.ListTools(ctx, mcp.ListToolsRequest{}) + Expect(err).NotTo(HaveOccurred()) + names := make([]string, 0, len(tools.Tools)) + for _, tool := range tools.Tools { + names = append(names, tool.Name) + } + Expect(names).To(ConsistOf("version", "contexts")) + + request := mcp.CallToolRequest{} + request.Params.Name = "version" + result, err := client.CallTool(ctx, request) + Expect(err).NotTo(HaveOccurred()) + Expect(result.IsError).To(BeFalse()) + Expect(result.StructuredContent).To(Equal(map[string]any{"version": "test"})) + + request.Params.Name = "whoami" + _, err = client.CallTool(ctx, request) + Expect(err).To(HaveOccurred()) + Expect(hiddenCalls.Load()).To(BeZero()) + }) + + It("rejects remote credentials with the wrong bearer or binding and after expiry or revocation", func(ctx SpecContext) { + remote := httptest.NewServer(callertools.RemoteHandler()) + DeferCleanup(remote.Close) + runtime := newRuntime("remote-auth-session", "remote") + DeferCleanup(runtime.Close) + issue := func(expiry time.Time) *api.CallerToolDelegation { + delegation, err := runtime.Endpoint().Delegate(ctx, api.CallerToolBinding{ + TaskID: "task-auth", Agent: "agent-auth", ExpiresAt: expiry, ToolNames: []string{"identity"}, + }) + Expect(err).NotTo(HaveOccurred()) + delegation.Endpoint = servedDelegation(remote.URL, delegation.Endpoint) + return delegation + } + + active := issue(time.Now().Add(time.Minute)) + DeferCleanup(active.Revoke) + invalidBearer := cloneEndpoint(active.Endpoint) + invalidBearer.Headers["Authorization"] = "Bearer invalid" + Expect(authenticatedStatus(invalidBearer)).To(Equal(http.StatusUnauthorized)) + wrongTask := cloneEndpoint(active.Endpoint) + wrongTask.Headers[callertools.TaskHeader] = "another-task" + Expect(authenticatedStatus(wrongTask)).To(Equal(http.StatusForbidden)) + wrongAgent := cloneEndpoint(active.Endpoint) + wrongAgent.Headers[callertools.AgentHeader] = "another-agent" + Expect(authenticatedStatus(wrongAgent)).To(Equal(http.StatusForbidden)) + + expiring := issue(time.Now().Add(25 * time.Millisecond)) + Eventually(func() int { return authenticatedStatus(expiring.Endpoint) }).Should(Equal(http.StatusUnauthorized)) + revoked := issue(time.Now().Add(time.Minute)) + revoked.Revoke() + Expect(authenticatedStatus(revoked.Endpoint)).To(Equal(http.StatusUnauthorized)) + }) + + It("returns terminal remote tool results for approval denial and broker failure", func(ctx SpecContext) { + remote := httptest.NewServer(callertools.RemoteHandler()) + DeferCleanup(remote.Close) + for _, test := range []struct { + name string + decision api.PermissionDecision + err error + message string + reason string + }{ + {name: "denied", decision: api.PermissionDecision{Message: "operator denied the call"}, message: "operator denied the call", reason: "approval_denied"}, + {name: "failed", err: errors.New("approval service unavailable"), message: "approval service unavailable", reason: "approval_failed"}, + } { + var calls atomic.Int32 + events := make(chan api.Event, 2) + audits := make(chan callertools.AuditEvent, 4) + runtime, err := callertools.New(callertools.Options{ + Definitions: []api.ToolDefinition{{ + Name: "version", DefaultPermission: api.ToolPolicyAsk, + Handler: func(context.Context, map[string]any) (any, error) { + calls.Add(1) + return "must not execute", nil + }, + }}, + SessionID: "remote-approval-" + test.name, + CanUseTool: func(_ context.Context, request api.PermissionRequest) (api.PermissionDecision, error) { + Expect(request.Delegated).To(BeTrue()) + Expect(request.ToolUseIDGenerated).To(BeTrue()) + return test.decision, test.err + }, + ObserveDelegatedTool: func(_ context.Context, event api.Event) error { + events <- event + return nil + }, + Audit: func(event callertools.AuditEvent) { audits <- event }, + }) + Expect(err).NotTo(HaveOccurred()) + delegation, err := runtime.Endpoint().Delegate(ctx, api.CallerToolBinding{ + TaskID: "task-" + test.name, Agent: "agent-1", ExpiresAt: time.Now().Add(time.Minute), + ToolNames: []string{"version"}, + }) + Expect(err).NotTo(HaveOccurred()) + client := authenticatedClient(ctx, servedDelegation(remote.URL, delegation.Endpoint)) + + request := mcp.CallToolRequest{} + request.Params.Name = "version" + result, err := client.CallTool(ctx, request) + Expect(err).NotTo(HaveOccurred()) + Expect(result.IsError).To(BeTrue()) + Expect(toolResultText(result)).To(ContainSubstring(test.message)) + Expect(calls.Load()).To(BeZero()) + var use, terminal api.Event + Eventually(events).Should(Receive(&use)) + Eventually(events).Should(Receive(&terminal)) + Expect(use.Kind).To(Equal(api.EventToolUse)) + Expect(terminal).To(MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(api.EventToolResult), "ToolCallID": Equal(use.ToolCallID), + "Success": BeFalse(), "Text": ContainSubstring(test.message), "Delegated": BeTrue(), + })) + Eventually(audits).Should(Receive(MatchFields(IgnoreExtras, Fields{ + "Action": Equal("call"), "Result": Equal("denied"), "Reason": Equal(test.reason), + }))) + + Expect(client.Close()).To(Succeed()) + delegation.Revoke() + Expect(runtime.Close()).To(Succeed()) + } + }) }) func authenticatedClient(ctx context.Context, endpoint api.CallerToolEndpoint) *mcpclient.Client { @@ -337,6 +502,29 @@ func authenticatedStatus(endpoint api.CallerToolEndpoint) int { return response.StatusCode } +func servedDelegation(serverURL string, endpoint api.CallerToolEndpoint) api.CallerToolEndpoint { + parsed, err := url.Parse(endpoint.URL) + Expect(err).NotTo(HaveOccurred()) + endpoint.URL = serverURL + parsed.RequestURI() + return endpoint +} + +func cloneEndpoint(endpoint api.CallerToolEndpoint) api.CallerToolEndpoint { + cloned := endpoint + cloned.Headers = make(map[string]string, len(endpoint.Headers)) + for name, value := range endpoint.Headers { + cloned.Headers[name] = value + } + return cloned +} + +func toolResultText(result *mcp.CallToolResult) string { + Expect(result.Content).NotTo(BeEmpty()) + text, ok := mcp.AsTextContent(result.Content[0]) + Expect(ok).To(BeTrue()) + return text.Text +} + func newRuntime(sessionID, marker string) *callertools.Runtime { runtime, err := callertools.New(callertools.Options{ Definitions: []api.ToolDefinition{{ diff --git a/pkg/aichat/execution.go b/pkg/aichat/execution.go index 88d82046..9f08b88a 100644 --- a/pkg/aichat/execution.go +++ b/pkg/aichat/execution.go @@ -99,6 +99,39 @@ func mergeExecutionEvents( deferred = deferred[:0] return true } + handleEvent := func(event api.Event) bool { + if event.Kind == api.EventToolUse { + if !send(event) { + return false + } + if askTools[event.Tool] { + awaiting[event.ToolCallID] = true + if approval, ok := pendingApprovals[event.ToolCallID]; ok { + if !send(approval) { + return false + } + delete(pendingApprovals, event.ToolCallID) + delete(awaiting, event.ToolCallID) + } + } + return true + } + if event.Kind == api.EventToolResult && awaiting[event.ToolCallID] { + if !send(event) { + return false + } + delete(awaiting, event.ToolCallID) + if len(awaiting) == 0 && !flush() { + return false + } + return true + } + if len(awaiting) > 0 { + deferred = append(deferred, event) + return true + } + return send(event) + } for provider != nil || (len(awaiting) > 0 && approvals != nil) { select { case <-ctx.Done(): @@ -108,6 +141,12 @@ func mergeExecutionEvents( approvals = nil continue } + if approval.Kind != api.EventPermission { + if !handleEvent(approval) { + return + } + continue + } if !awaiting[approval.ToolCallID] { pendingApprovals[approval.ToolCallID] = approval continue @@ -124,27 +163,7 @@ func mergeExecutionEvents( provider = nil continue } - if event.Kind == api.EventToolUse { - if !send(event) { - return - } - if askTools[event.Tool] { - awaiting[event.ToolCallID] = true - if approval, ok := pendingApprovals[event.ToolCallID]; ok { - if !send(approval) { - return - } - delete(pendingApprovals, event.ToolCallID) - delete(awaiting, event.ToolCallID) - } - } - continue - } - if len(awaiting) > 0 { - deferred = append(deferred, event) - continue - } - if !send(event) { + if !handleEvent(event) { return } } diff --git a/pkg/aichat/execution_authority_ginkgo_test.go b/pkg/aichat/execution_authority_ginkgo_test.go index 2359ea7c..6a5002e4 100644 --- a/pkg/aichat/execution_authority_ginkgo_test.go +++ b/pkg/aichat/execution_authority_ginkgo_test.go @@ -209,6 +209,85 @@ var _ = Describe("Authoritative aichat execution", func() { Expect(execution.closed).To(BeTrue()) }) + It("streams a delegated caller-tool lifecycle reconstructed from its MCP call", func() { + store := aichat.NewMemoryThreadStore() + thread, err := store.Create(context.Background(), "Delegated") + Expect(err).NotTo(HaveOccurred()) + execution := &fakeExecution{ + events: make(chan api.Event), + endpoint: &api.CallerToolEndpoint{ + Name: "captain", URL: "http://127.0.0.1:43210/mcp", + Headers: map[string]string{"Authorization": "Bearer secret"}, + }, + } + provider := &fakeStreamingProvider{ + backend: api.BackendCodexAgent, + execute: func(context.Context, api.Spec) (<-chan api.Event, error) { + providerEvents := make(chan api.Event) + go func() { + execution.events <- api.Event{ + Kind: api.EventToolUse, Tool: "version", ToolCallID: "mcp-call-1", + Input: map[string]any{}, Delegated: true, + } + execution.events <- api.Event{ + Kind: api.EventPermission, Tool: "version", ToolCallID: "mcp-call-1", + ApprovalID: "approval-mcp-call-1", Input: map[string]any{}, Delegated: true, + } + execution.events <- api.Event{ + Kind: api.EventToolResult, Tool: "version", ToolCallID: "mcp-call-1", + Text: `{"version":"test"}`, Success: true, Delegated: true, + } + execution.events <- api.Event{ + Kind: api.EventToolUse, Tool: "version", ToolCallID: "mcp-call-2", + Input: map[string]any{}, Delegated: true, + } + execution.events <- api.Event{ + Kind: api.EventToolResult, Tool: "version", ToolCallID: "mcp-call-2", + Text: "approval service unavailable", Success: false, Delegated: true, + } + providerEvents <- api.Event{Kind: api.EventResult, Success: true, SessionID: "remote-session-1"} + close(providerEvents) + }() + return providerEvents, nil + }, + } + service := aichat.NewService(aichat.ServiceOptions{ + Resolver: &fakeResolver{provider: provider}, Threads: aichat.FixedThreadStore(store), + Authority: &fakeExecutionAuthority{execution: execution}, + Tools: aichat.StaticToolProvider([]api.ToolDefinition{{ + Name: "version", DefaultPermission: api.ToolPolicyAsk, + Handler: func(context.Context, map[string]any) (any, error) { return nil, nil }, + }}), + }) + + response := httptest.NewRecorder() + service.Handler().ServeHTTP(response, requestJSON(http.MethodPost, "/api/chat", aichat.ChatRequest{ + ID: thread.ID, ThreadID: thread.ID, Trigger: "submit-message", + Runtime: &api.Model{Name: "codex", Backend: api.BackendCodexAgent}, + Messages: []aichat.UIMessage{{ + ID: "user-message-delegated", Role: "user", + Parts: []aichat.UIPart{{Type: "text", Text: "Check the version remotely"}}, + }}, + })) + + parts := decodedDataLines(response.Body.String()) + Expect(response.Code).To(Equal(http.StatusOK), response.Body.String()) + Expect(partTypes(parts)).To(Equal([]string{ + "start", "start-step", "tool-input-available", "tool-approval-request", + "tool-output-available", "tool-input-available", "tool-output-error", + "data-result", "finish-step", "finish", + })) + Expect(execution.observed).To(ContainElements( + MatchFields(IgnoreExtras, Fields{"Kind": Equal(api.EventToolUse), "ToolCallID": Equal("mcp-call-1")}), + MatchFields(IgnoreExtras, Fields{"Kind": Equal(api.EventPermission), "ToolCallID": Equal("mcp-call-1")}), + MatchFields(IgnoreExtras, Fields{"Kind": Equal(api.EventToolResult), "ToolCallID": Equal("mcp-call-1")}), + MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(api.EventToolResult), "ToolCallID": Equal("mcp-call-2"), + "Success": BeFalse(), "Text": Equal("approval service unavailable"), + }), + )) + }) + It("streams API-provider approvals without waiting for agent caller-tool events", func() { store := aichat.NewMemoryThreadStore() thread, err := store.Create(context.Background(), "Accounts") diff --git a/pkg/aichat/execution_database.go b/pkg/aichat/execution_database.go index 541e12e5..2187a9f9 100644 --- a/pkg/aichat/execution_database.go +++ b/pkg/aichat/execution_database.go @@ -131,7 +131,8 @@ func (e *databaseExecution) startCallerTools(ctx context.Context, backend api.Ba var credentialID uuid.UUID runtime, err := callertools.New(callertools.Options{ Definitions: e.definitions, SessionID: e.session.ID.String(), - ApprovalTimeout: callerToolApprovalTimeout, + ApprovalTimeout: callerToolApprovalTimeout, + ObserveDelegatedTool: e.emitCallerToolEvent, ValidateCredential: func(ctx context.Context) error { if credentialID == uuid.Nil { return fmt.Errorf("caller-tool credential has not been issued") @@ -172,7 +173,7 @@ func (e *databaseExecution) requestApproval( credentialID uuid.UUID, request api.PermissionRequest, ) (api.PermissionDecision, error) { - if request.ToolUseIDGenerated { + if request.ToolUseIDGenerated && !request.Delegated { toolUseID, err := e.claimProviderToolUse(ctx, request) if err != nil { return api.PermissionDecision{}, err @@ -200,11 +201,7 @@ func (e *databaseExecution) requestApproval( return decision, errors.Join(err, restoreErr) } -func (e *databaseExecution) emitApproval(ctx context.Context, approvalID uuid.UUID, request api.PermissionRequest) error { - event := api.Event{ - Kind: api.EventPermission, Tool: request.Tool, - ToolCallID: request.ToolUseID, ApprovalID: approvalID.String(), Input: request.Input, - } +func (e *databaseExecution) emitCallerToolEvent(ctx context.Context, event api.Event) error { select { case e.events <- event: return nil @@ -213,6 +210,15 @@ func (e *databaseExecution) emitApproval(ctx context.Context, approvalID uuid.UU } } +func (e *databaseExecution) emitApproval(ctx context.Context, approvalID uuid.UUID, request api.PermissionRequest) error { + event := api.Event{ + Kind: api.EventPermission, Tool: request.Tool, + ToolCallID: request.ToolUseID, ApprovalID: approvalID.String(), Input: request.Input, + Delegated: request.Delegated, + } + return e.emitCallerToolEvent(ctx, event) +} + func (e *databaseExecution) waitForApproval( ctx context.Context, requestID uuid.UUID, @@ -267,7 +273,9 @@ func (e *databaseExecution) Observe(ctx context.Context, event api.Event) (api.E } switch event.Kind { case api.EventToolUse: - e.rememberProviderToolUse(event) + if !event.Delegated { + e.rememberProviderToolUse(event) + } return event, nil case api.EventPermission: if event.ApprovalID != "" { diff --git a/pkg/aichat/execution_database_integration_test.go b/pkg/aichat/execution_database_integration_test.go index 7cddc10d..9941150c 100644 --- a/pkg/aichat/execution_database_integration_test.go +++ b/pkg/aichat/execution_database_integration_test.go @@ -2,13 +2,21 @@ package aichat_test import ( "context" + "encoding/pem" "errors" + "net/http" + "net/http/httptest" + "os" + "path/filepath" "sync/atomic" "time" + "github.com/flanksource/captain/pkg/ai/callertools" "github.com/flanksource/captain/pkg/aichat" "github.com/flanksource/captain/pkg/api" "github.com/flanksource/captain/pkg/database" + "github.com/flanksource/captain/pkg/gitagent" + "github.com/flanksource/clicky/text" "github.com/flanksource/commons-db/dbtest" "github.com/google/uuid" mcpclient "github.com/mark3labs/mcp-go/client" @@ -18,6 +26,7 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + . "github.com/onsi/gomega/gstruct" ) var _ = Describe("Database execution authority", func() { @@ -108,6 +117,153 @@ var _ = Describe("Database execution authority", func() { Expect(credential.RevokedAt).NotTo(BeNil()) }) + It("approves and executes an authenticated delegated tool without a local provider tool-use event", func(ctx SpecContext) { + testDB := dbtest.ForGinkgo(dbtest.Options{Name: "captain_aichat_delegated_execution"}) + db, err := database.Open(ctx, database.WithDSN(testDB.DSN()), database.WithMigrations()) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(db.Close) + authority, err := aichat.NewDatabaseExecutionAuthority(db) + Expect(err).NotTo(HaveOccurred()) + var calls atomic.Int32 + threadID := uuid.NewString() + execution, err := authority.Begin(ctx, aichat.ExecutionRequest{ + ThreadID: threadID, RequestID: "request-delegated-1", Title: "Delegated", + Spec: api.Spec{Model: api.Model{ + Name: "codex", Backend: api.BackendCodexAgent, + }.Capabilities()}, + Definitions: []api.ToolDefinition{{ + Name: "version", DefaultPermission: api.ToolPolicyAsk, + Handler: func(context.Context, map[string]any) (any, error) { + calls.Add(1) + return map[string]any{"version": "test"}, nil + }, + }}, + }) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(execution.Close) + supervisor := httptest.NewTLSServer(callertools.RemoteHandler()) + DeferCleanup(supervisor.Close) + endpoint := execution.CallerTools() + Expect(endpoint).NotTo(BeNil()) + delegation, err := endpoint.Delegate(ctx, api.CallerToolBinding{ + TaskID: "task-delegated-1", Agent: "agent-1", ExpiresAt: time.Now().Add(time.Minute), + ToolNames: []string{"version"}, + }) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(delegation.Revoke) + + sidecarRoot := GinkgoT().TempDir() + sidecar := httptest.NewUnstartedServer(nil) + sidecarHandler, err := gitagent.NewCallerToolProxy(gitagent.CallerToolProxyConfig{ + Root: sidecarRoot, EndpointURL: "https://" + sidecar.Listener.Addr().String(), + SupervisorURL: supervisor.URL, SupervisorCAPath: testServerCAPath(supervisor), + Agent: "agent-1", DefaultRunner: true, + IdentifySupervisor: func(request *http.Request) (string, error) { + if request.Header.Get("Authorization") != "Bearer dispatch-token" { + return "", errors.New("invalid supervisor credential") + } + return "supervisor", nil + }, + }) + Expect(err).NotTo(HaveOccurred()) + sidecar.Config.Handler = sidecarHandler + sidecar.StartTLS() + DeferCleanup(sidecar.Close) + target := gitagent.TransportTarget{ + URL: sidecar.URL, Token: text.NewSensitiveString("dispatch-token"), + CAPath: testServerCAPath(sidecar), + } + Expect(gitagent.RegisterCallerTools(ctx, target, gitagent.CallerToolGrant{ + Task: "task-delegated-1", Agent: "agent-1", Endpoint: delegation.Endpoint, + ExpiresAt: time.Now().Add(time.Minute), + })).To(Succeed()) + remoteEndpoint, err := gitagent.LoadCallerToolEndpoint(sidecarRoot, "task-delegated-1") + Expect(err).NotTo(HaveOccurred()) + client := executionMCPClientWithHTTP(ctx, *remoteEndpoint, sidecar.Client()) + DeferCleanup(client.Close) + type callOutcome struct { + result *mcp.CallToolResult + err error + } + outcomes := make(chan callOutcome, 1) + go func() { + request := mcp.CallToolRequest{} + request.Params.Name = "version" + result, callErr := client.CallTool(ctx, request) + outcomes <- callOutcome{result: result, err: callErr} + }() + + var use api.Event + Eventually(execution.Events()).Should(Receive(&use)) + Expect(use).To(MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(api.EventToolUse), "Tool": Equal("version"), + })) + Expect(use.ToolCallID).To(HavePrefix("mcp_")) + var approval api.Event + Eventually(execution.Events()).Should(Receive(&approval)) + Expect(approval).To(MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(api.EventPermission), "Tool": Equal("version"), + "ToolCallID": Equal(use.ToolCallID), + })) + Expect(calls.Load()).To(BeZero()) + continuation, err := authority.ResolveToolApproval(ctx, aichat.ToolApprovalResolution{ + ThreadID: threadID, ApprovalID: approval.ApprovalID, Approved: true, + }) + Expect(err).NotTo(HaveOccurred()) + Expect(continuation).To(BeNil()) + var terminal api.Event + Eventually(execution.Events()).Should(Receive(&terminal)) + Expect(terminal).To(MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(api.EventToolResult), "Tool": Equal("version"), + "ToolCallID": Equal(use.ToolCallID), "Success": BeTrue(), + })) + var outcome callOutcome + Eventually(outcomes).Should(Receive(&outcome)) + Expect(outcome.err).NotTo(HaveOccurred()) + Expect(outcome.result.IsError).To(BeFalse()) + Expect(outcome.result.StructuredContent).To(Equal(map[string]any{"version": "test"})) + Expect(calls.Load()).To(Equal(int32(1))) + + go func() { + request := mcp.CallToolRequest{} + request.Params.Name = "version" + result, callErr := client.CallTool(ctx, request) + outcomes <- callOutcome{result: result, err: callErr} + }() + var deniedUse, deniedApproval api.Event + Eventually(execution.Events()).Should(Receive(&deniedUse)) + Eventually(execution.Events()).Should(Receive(&deniedApproval)) + Expect(deniedUse.Kind).To(Equal(api.EventToolUse)) + Expect(deniedApproval).To(MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(api.EventPermission), "ToolCallID": Equal(deniedUse.ToolCallID), + })) + continuation, err = authority.ResolveToolApproval(ctx, aichat.ToolApprovalResolution{ + ThreadID: threadID, ApprovalID: deniedApproval.ApprovalID, Approved: false, + Reason: "operator denied remote version", + }) + Expect(err).NotTo(HaveOccurred()) + Expect(continuation).To(BeNil()) + Eventually(execution.Events()).Should(Receive(&terminal)) + Expect(terminal).To(MatchFields(IgnoreExtras, Fields{ + "Kind": Equal(api.EventToolResult), "ToolCallID": Equal(deniedUse.ToolCallID), + "Success": BeFalse(), "Text": Equal("operator denied remote version"), + })) + Eventually(outcomes).Should(Receive(&outcome)) + Expect(outcome.err).NotTo(HaveOccurred()) + Expect(outcome.result.IsError).To(BeTrue()) + Expect(outcome.result.Content).NotTo(BeEmpty()) + text, ok := mcp.AsTextContent(outcome.result.Content[0]) + Expect(ok).To(BeTrue()) + Expect(text.Text).To(Equal("operator denied remote version")) + Expect(calls.Load()).To(Equal(int32(1))) + + _, err = execution.Observe(ctx, api.Event{ + Kind: api.EventResult, Success: true, SessionID: "remote-provider-session-1", + }) + Expect(err).NotTo(HaveOccurred()) + Expect(execution.Close(ctx)).To(Succeed()) + }) + It("creates distinct prompt runs for sequential turn identities and rejects a replay", func(ctx SpecContext) { testDB := dbtest.ForGinkgo(dbtest.Options{Name: "captain_aichat_sequential_turns"}) db, err := database.Open(ctx, database.WithDSN(testDB.DSN()), database.WithMigrations()) @@ -451,7 +607,15 @@ var _ = Describe("Database execution authority", func() { }) func executionMCPClient(ctx context.Context, endpoint api.CallerToolEndpoint) *mcpclient.Client { - channel, err := transport.NewStreamableHTTP(endpoint.URL, transport.WithHTTPHeaders(endpoint.Headers)) + return executionMCPClientWithHTTP(ctx, endpoint, nil) +} + +func executionMCPClientWithHTTP(ctx context.Context, endpoint api.CallerToolEndpoint, httpClient *http.Client) *mcpclient.Client { + options := []transport.StreamableHTTPCOption{transport.WithHTTPHeaders(endpoint.Headers)} + if httpClient != nil { + options = append(options, transport.WithHTTPBasicClient(httpClient)) + } + channel, err := transport.NewStreamableHTTP(endpoint.URL, options...) Expect(err).NotTo(HaveOccurred()) client := mcpclient.NewClient(channel) Expect(client.Start(ctx)).To(Succeed()) @@ -462,3 +626,12 @@ func executionMCPClient(ctx context.Context, endpoint api.CallerToolEndpoint) *m Expect(err).NotTo(HaveOccurred()) return client } + +func testServerCAPath(server *httptest.Server) string { + certificate := server.Certificate() + Expect(certificate).NotTo(BeNil()) + payload := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certificate.Raw}) + path := filepath.Join(GinkgoT().TempDir(), "ca.pem") + Expect(os.WriteFile(path, payload, 0o600)).To(Succeed()) + return path +} diff --git a/pkg/api/runtime_config.go b/pkg/api/runtime_config.go index e35b0024..8f7438e0 100644 --- a/pkg/api/runtime_config.go +++ b/pkg/api/runtime_config.go @@ -21,7 +21,11 @@ type PermissionRequest struct { Input map[string]any ToolUseID string ToolUseIDGenerated bool - SessionID string + // Delegated identifies a call authenticated by a task-scoped remote + // capability. Its MCP request is the authoritative tool-use observation; + // there is no local provider event to correlate with a generated ID. + Delegated bool `json:"-"` + SessionID string } // PermissionDecision is the answer to a PermissionRequest. On Allow the tool runs diff --git a/pkg/api/runtime_event.go b/pkg/api/runtime_event.go index e3a29b23..66fab992 100644 --- a/pkg/api/runtime_event.go +++ b/pkg/api/runtime_event.go @@ -60,6 +60,9 @@ type Event struct { // (the call) and EventToolResult (its complete output). Backends that stream // output incrementally accumulate it and emit a single EventToolResult. ToolCallID string + // Delegated marks a lifecycle event reconstructed by the supervisor from an + // authenticated remote MCP call rather than emitted by its local provider. + Delegated bool `json:"-"` // ApprovalID is the durable captain_turn_requests UUID associated with an // EventPermission. It is distinct from the provider's tool-call ID. ApprovalID string From d800c7ed996c47494de6990fa51f851dd4e24571 Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Mon, 31 Aug 2026 17:51:57 +0000 Subject: [PATCH 6/6] fix(git-agent): address caller-tool review findings Keep delegated capabilities alive across dispatch setup and preserve completed tool outcomes when terminal event delivery fails, avoiding misleading retries of non-idempotent handlers. Log failures to remove transient caller-tool credentials so operators can detect secrets that require startup cleanup. Amp-Thread-ID: https://ampcode.com/threads/T-01a0580d-57bd-769a-8f96-d01d5ba28789 --- pkg/ai/callertools/runtime.go | 2 +- pkg/gitagent/callertools.go | 8 ++++++-- pkg/sandbox/adapter/gitagent.go | 6 +++++- 3 files changed, 12 insertions(+), 4 deletions(-) diff --git a/pkg/ai/callertools/runtime.go b/pkg/ai/callertools/runtime.go index eda14971..27e6ea61 100644 --- a/pkg/ai/callertools/runtime.go +++ b/pkg/ai/callertools/runtime.go @@ -467,7 +467,7 @@ func (r *Runtime) handler(definition api.ToolDefinition) server.ToolHandlerFunc Text: string(encoded), Success: true, Delegated: true, }); err != nil { r.auditCall(ctx, definition.Name, "error", "event_delivery_failed") - return mcp.NewToolResultErrorf("record delegated caller-tool result: %v", err), nil + return result, nil } } r.auditCall(ctx, definition.Name, "allowed", "") diff --git a/pkg/gitagent/callertools.go b/pkg/gitagent/callertools.go index b6c4997e..8e49c14e 100644 --- a/pkg/gitagent/callertools.go +++ b/pkg/gitagent/callertools.go @@ -304,7 +304,9 @@ func (proxy *callerToolProxy) revoke(w http.ResponseWriter, request *http.Reques if session != nil && session.revoked.CompareAndSwap(false, true) { proxy.log("git-agent caller-tool grant revoked task=%s agent=%s", session.task, session.agent) } - _ = removeCallerToolSecret(proxy.root, task) + if err := removeCallerToolSecret(proxy.root, task); err != nil { + proxy.log("git-agent caller-tool secret cleanup failed task=%s: %v", task, err) + } w.WriteHeader(http.StatusNoContent) } @@ -319,7 +321,9 @@ func (proxy *callerToolProxy) expire(session *callerToolSession) { proxy.mu.Unlock() proxy.log("git-agent caller-tool grant expired task=%s agent=%s", session.task, session.agent) if removed { - _ = removeCallerToolSecret(proxy.root, session.task) + if err := removeCallerToolSecret(proxy.root, session.task); err != nil { + proxy.log("git-agent caller-tool secret cleanup failed task=%s agent=%s: %v", session.task, session.agent, err) + } } } } diff --git a/pkg/sandbox/adapter/gitagent.go b/pkg/sandbox/adapter/gitagent.go index 42bf898e..cdb4ee0a 100644 --- a/pkg/sandbox/adapter/gitagent.go +++ b/pkg/sandbox/adapter/gitagent.go @@ -89,7 +89,9 @@ func (g *gitAgentSandbox) Execute(ctx context.Context, spec api.Spec) (*api.Resp if err != nil { return nil, err } - expiresAt := time.Now().Add(target.waitTimeout) + // AwaitOutcome starts its full timeout only after mailbox setup and the + // dispatch push, so the capability needs a small head-start margin. + expiresAt := time.Now().Add(target.waitTimeout + callerToolExpiryGrace) if deadline, ok := ctx.Deadline(); ok && deadline.Before(expiresAt) { expiresAt = deadline } @@ -342,6 +344,8 @@ const ( servedReposDir = "repos" ) +const callerToolExpiryGrace = 5 * time.Minute + // DefaultWaitTimeout bounds how long a dispatch waits for its verdict. A // relocating sandbox blocks on a remote agent doing real work, so this is // sized for that rather than for a model call.