diff --git a/pkg/ai/callertools/runtime.go b/pkg/ai/callertools/runtime.go index f85643ae..27e6ea61 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,10 @@ 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) + // 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 } @@ -54,14 +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 + 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 @@ -71,11 +100,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()) { @@ -104,25 +155,28 @@ 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, - 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. mcpServer := server.NewMCPServer( "captain-caller-tools", "1.0.0", server.WithToolCapabilities(false), server.WithToolFilter(runtime.filterTools), - server.WithInputSchemaValidation(), ) for _, definition := range definitions { runtime.definitions[definition.Name] = definition @@ -140,8 +194,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 +205,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 +220,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 +265,103 @@ 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, + // 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, + 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,64 +370,119 @@ 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() + capability, delegated := ctx.Value(delegationContextKey{}).(*delegatedCapability) + if delegated { + 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 } + 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 { + return fail(err.Error(), "denied", "invalid_input") + } if definition.NeedsApproval() { if r.canUseTool == nil { - 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 { - 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" } - return mcp.NewToolResultError(message), nil + return fail(message, "denied", "approval_denied") } if decision.UpdatedInput != nil { input = decision.UpdatedInput } } - if err := r.validateActive(callCtx); err != nil { - return mcp.NewToolResultError(err.Error()), nil - } if err := r.validateInput(definition.Name, input); err != nil { - return mcp.NewToolResultError(err.Error()), nil + return fail(err.Error(), "denied", "invalid_input") + } + if err := r.validateActive(callCtx); err != nil { + return fail(err.Error(), "denied", "capability_inactive") } output, err := definition.Handler(callCtx, input) if err != nil { - return mcp.NewToolResultError(err.Error()), nil + return fail(err.Error(), "error", "handler_failed") } result, err := mcp.NewToolResultJSON(output) if err != nil { - 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 result, nil + } } + r.auditCall(ctx, definition.Name, "allowed", "") return result, nil } } -func (r *Runtime) authorize(next http.Handler) http.Handler { +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) if err != nil || !net.ParseIP(host).IsLoopback() { @@ -282,6 +505,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 +544,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 +555,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/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/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/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/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/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 826a9d88..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,6 +247,11 @@ func (s *Service) handleChat(w http.ResponseWriter, request *http.Request) { config.SessionID = spec.SessionID config.CaptainSessionID = chat.ThreadID config.Tools = definitions + 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 { http.Error(w, err.Error(), http.StatusServiceUnavailable) 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/api/runtime_config.go b/pkg/api/runtime_config.go index e63fc7c0..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 @@ -48,8 +52,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/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 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..fdfa52c7 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,12 @@ 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, + callerTools *api.CallerToolEndpoint, +) error { providerOpts := AIProviderOptions{ ModelFlags: aiflags.ModelFlags{Model: payload.Model, Backend: payload.Backend, Effort: string(payload.Effort)}, Sandbox: "none", @@ -88,6 +103,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_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/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/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), + })); +} diff --git a/pkg/gitagent/callertools.go b/pkg/gitagent/callertools.go new file mode 100644 index 00000000..8e49c14e --- /dev/null +++ b/pkg/gitagent/callertools.go @@ -0,0 +1,638 @@ +// 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/http/httputil" + "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() + 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() + http.Error(w, "store caller-tool task secret", http.StatusInternalServerError) + return + } + 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)) + 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, "" + 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) + }, + } + reverseProxy.ServeHTTP(w, request) +} + +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) + } + 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) +} + +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) + if removed { + 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) + } + } + } +} + +// 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) { + name, err := callerToolSecretName(task) + if err != nil { + return nil, err + } + 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 must not be accessible by group or other users") + } + payload, err := root.ReadFile(name) + if err != nil { + return nil, err + } + if err := root.Remove(name); 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 +} + +// 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 + } + 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 + } + created = false + return nil +} + +func removeCallerToolSecret(root, task string) error { + 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 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) { + 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 := store.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 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..cdb4ee0a 100644 --- a/pkg/sandbox/adapter/gitagent.go +++ b/pkg/sandbox/adapter/gitagent.go @@ -63,6 +63,61 @@ 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 + } + // 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 + } + 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 +159,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 +232,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) { @@ -280,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.